RabbitMQ 3节点Docker集群经HAProxy代理 节点重启后SAC队列消费者无法连接

• 集群部署方式：基于Docker搭建3节点RabbitMQ集群，集群前置HAProxy代理
• 版本明细：
  • RabbitMQ: 3.8.11
  • Erlang: 23.2.3
  • Spring-amqp: 1.7.3

业务应用基于Spring Boot 1.5.4开发，共配置3个队列：

• 队列1：exclusive（排他）、durable（持久化）、auto-delete=false
• 队列2：SAC（单活跃消费者）、durable（持久化）、auto-delete=false
• 队列3：classic（经典）、durable（持久化）、auto-delete=false

队列匹配策略见下图：

1. 应用首次启动时，所有队列均可正常完成注册；
2. 随机下线任意RabbitMQ节点时，若该节点为队列master节点，会触发镜像队列选主流程，由镜像节点升任新master，此时业务运行无异常；
3. 待下线节点重新启动后，应用日志开始抛出如下异常：

Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - failed to perform operation on queue 'single-active-consumer-message-queue' in vhost '/dev' due to timeout, class-id=50, method-id=10)
    at com.rabbitmq.utility.ValueOrException.getValue(ValueOrException.java:66) ~[amqp-client-4.0.2.jar!/:4.0.2]
    at com.rabbitmq.utility.BlockingValueOrException.uninterruptibleGetValue(BlockingValueOrException.java:32) ~[amqp-client-4.0.2.jar!/:4.0.2]
    at com.rabbitmq.client.impl.AMQChannel$BlockingRpcContinuation.getReply(AMQChannel.java:366) ~[amqp-client-4.0.2.jar!/:4.0.2]
    at com.rabbitmq.client.impl.AMQChannel.privateRpc(AMQChannel.java:229) ~[amqp-client-4.0.2.jar!/:229]
    at com.rabbitmq.client.impl.AMQChannel.exnWrappingRpc(AMQChannel.java:117) ~[amqp-client-4.0.2.jar!/:117]
    ... 25 common frames omitted

应用连续尝试重连3次后，最终输出如下日志：

2021-03-18 17:08:55.487 ERROR 1 --- [cTaskExecutor-4] o.s.a.r.l.SimpleMessageListenerContainer : Stopping container from aborted consumer
2021-03-18 17:08:55.487  INFO 1 --- [cTaskExecutor-4] o.s.a.r.l.SimpleMessageListenerContainer : Waiting for workers to finish.
2021-03-18 17:08:55.487  INFO 1 --- [cTaskExecutor-4] o.s.a.r.l.SimpleMessageListenerContainer : Successfully waited for workers to finish.

对应的RabbitMQ管理控制台异常表现见下图：

• CachingConnectionFactory采用HAProxy提供的基础连接参数配置
• 自定义的rabbitListenerContainerFactory代码实现如下：

@Bean
    protected SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory(ConnectionFactory connectionFactory, RetryOperationsInterceptor retryAdvice) {
        SimpleRabbitListenerContainerFactory containerFactory = new SimpleRabbitListenerContainerFactory();
        containerFactory.setConnectionFactory(connectionFactory);
        containerFactory.setDefaultRequeueRejected(false);
        containerFactory.setAdviceChain(retryAdvice);
        containerFactory.setMessageConverter(new Jackson2JsonMessageConverter());
        return containerFactory;
    }

节点重启后SAC队列的消费者无法正常建立连接，需要定位问题根因并提供可行解决方案。

内容的提问来源于stack exchange，提问作者Shuki