heshiyingx commented on issue #1033:
URL:
https://github.com/apache/rocketmq-client-go/issues/1033#issuecomment-1577151957
> > 看错误是消费端挂了，我怀疑你消费端调用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)
> }
> }
> ```
在这里看是没问题的，你再确认一下，在运行的这个进程里(这个项目)，在其他地方有没有创建的有消费者，然后调用了shutdown方法的？
--
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]
