rabbitMq 报错 o.s.a.r.c.CachingConnectionFactory: Channel shutdown: channel error； protocol method_if the broker stops and is restarted while the con

无意中发现测试服务器上会打印出这样的日志

2024-01-04 10:43:16.114 [AMQP Connection 10.150.30.45:5672] ERROR [server,test] [7] o.s.a.r.c.CachingConnectionFactory : Channel shutdown: channel error; protocol method: #method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - delivery acknowledgement on channel 64 timed out. Timeout value used: 1800000 ms. This timeout value can be configured, see consumers doc guide to learn more, class-id=0, method-id=0)
2024-01-04 10:43:16.336 [org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#2-130] INFO [new_dialer_server,test] [7] o.s.a.r.l.SimpleMessageListenerContainer : Restarting Consumer@21e75002: tags=[[amq.ctag-HKsBi1Xg9Afr7xYeMG20Gw]], channel=Cached Rabbit Channel: AMQChannel(amqp://dialog@10.150.30.45:5672/dialog_vhost,64), conn: Proxy@5ab956d7 Shared Rabbit Connection: SimpleConnection@2892d68 [delegate=amqp://dialog@10.150.30.45:5672/dialog_vhost, localPort= 50088], acknowledgeMode=MANUAL local queue size=0
2024-01-04 10:43:16.356 [org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#2-131] INFO [server,test] [7] o.s.amqp.rabbit.core.RabbitAdmin : Auto-declaring a non-durable or auto-delete Exchange (call_center) durable:false, auto-delete:false. It will be deleted by the broker if it shuts down, and can be redeclared by closing and reopening the connection.
2024-01-04 10:43:16.357 [org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#2-131] INFO [server,test] [7] o.s.amqp.rabbit.core.RabbitAdmin : Auto-declaring a non-durable or auto-delete Exchange (call_center) durable:false, auto-delete:false. It will be deleted by the broker if it shuts down, and can be redeclared by closing and reopening the connection.
[org.springframework.amqp.rabbit.RabbitListenerEndpointContainer#2-131] INFO [server,test] [7] o.s.amqp.rabbit.core.RabbitAdmin : Auto-declaring a non-durable, auto-delete, or exclusive Queue (queue.svr_state) durable:false, auto-delete:true, exclusive:false. It will be redeclared if the broker stops and is restarted while the connection factory is alive, but all messages will be lost.

一开始以为是项目重启时候，消费者跟着重新启动呢，就没在意。知道我在管理界面看到了unacked列有消息，背后一凉，事情绝没有这么简单…

趁着测试还没发现问题，赶紧排查，争取做到一个神不知鬼不觉的修复问题。

Channel shutdown: channel error; protocol method:
#method<channel.close>(reply-code=406, reply-text=PRECONDITION_FAILED - delivery acknowledgement on channel 64 timed out. Timeout value used: 1800000 ms. This timeout value can be configured, see consumers
doc guide to learn more, class-id=0, method-id=0)

这段错误的大概意思就是 消息送达确认传递超时了，我们都知道mq的ack机制，如果acknowledge-mode设置成manual(手动的)，就需要在消费者消费完消息之后，手动提交ack，告知通道，此消息已经被消费，可以删除了。

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: guest
    password: guest
    connection-timeout: 15000
    virtual-host: dialog_vhost
    listener:
      simple:
        acknowledge-mode: manual
1
2
3
4
5
6
7
8
9
10
11

看看代码里的端倪

@RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "queue.svr_state", durable = "false", exclusive = "false", autoDelete = "true"),
            exchange = @Exchange(value = RabbitmqConfig.CALL_CENTER_EXCHANGE, type = ExchangeTypes.TOPIC, durable = "false"),
            key = RabbitmqConfig.ROUTING_KEY
    ))
    @RabbitHandler
    public void svrState(Channel channel, Message message, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws Exception {
        log.info("------->消费者开始消费消息-------");
        JSONObject message = JSON.parseObject(message.getBody(), JSONObject.class);
        String svrPhone = String.valueOf(message.get("svr_phone"));
        if (StringUtils.isBlank(svrPhone) {
            return;
        }
        if (isContainsNumber(svrPhone)) {
            sendToBrowser(JSONObject.toJSONString(message), svrPhone);
        } 
        channel.basicAck(tag, false);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

最后一行确实手动提交确认了，但是当svrPhone为空的时候，会直接return，此时并不会提交确认，并且会将该消息放到Unacked里面。
解决办法就是在return之前，加上channel.basicAck(tag, false);就可以了。