Jimmy01010 commented on issue #1033:
URL:
https://github.com/apache/rocketmq-client-go/issues/1033#issuecomment-1574706727
> 看错误是消费端挂了，我怀疑你消费端调用Shutdown了，但是程序没有关闭，能否把你消费端的代码贴出来看一下
```
func ConfigConsumer() {
c, err := rocketmq.NewPushConsumer(
consumer.WithConsumerModel(consumer.Clustering),
consumer.WithNameServer([]string{global.ServerConfig.RocketMQ.Address}),
consumer.WithGroupName("config_consumer_"+global.ServerConfig.Name),
// consumer.WithGroupName("config_consumer_test1"),
consumer.WithConsumeFromWhere(consumer.ConsumeFromFirstOffset),
consumer.WithConsumerOrder(true),
consumer.WithConsumerPullTimeout(time.Second*60),
)
if err != nil {
panic("ConfigConsumer NewPushConsumer err:" + err.Error())
}
if err := c.Subscribe(global.ServerConfig.RocketMQ.Topic,
consumer.MessageSelector{},
func(ctx context.Context, msgs ...*primitive.MessageExt)
(consumer.ConsumeResult, error) {
for i := range msgs {
err := DealMessage(msgs[i].Body)
if err != nil {
zap.S().Errorf("处理消息错误 %v %v", msgs[i],
err)
return consumer.ConsumeRetryLater, nil
}
}
return consumer.ConsumeSuccess, nil
}); err != nil {
zap.S().Errorf("MQ:读取消息失败")
}
//开始消费消息
err = c.Start()
if err != nil {
zap.S().Errorf("ConfigConsumer Start Consumer失败: %v", err)
}
quit := make(chan os.Signal)
signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM) //nolint:govet
<-quit
err = c.Shutdown() //nolint:errcheck
if err != nil {
zap.S().Errorf("ConfigConsumer Shutdown 失败: %v", err)
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
