测试中如何触发RabbitMQ关闭信号并触发ShutdownListener？

首先明确：ShutdownListener仅监听RabbitMQ连接或通道的关闭事件，你遇到的「unacked消息超时导致消费者关闭」属于消费者被取消的场景，不会触发该监听器——这类场景需用ConsumerShutdownListener监听消费者取消事件。

针对你的ShutdownListener未触发的问题，下面给出可复现的触发场景及排查要点：

一、触发ShutdownListener的有效场景 #

1. 连接级关闭（触发hardError分支） #

• 直接停止RabbitMQ服务进程
• 用防火墙阻断客户端与RabbitMQ节点的网络连接
• 在RabbitMQ控制台手动断开目标连接
• 代码中调用connection.abort()（强制关闭连接，区别于应用主动优雅关闭）

2. 通道级关闭（触发else分支） #

• 违反RabbitMQ协议操作：比如重复声明同名但属性不一致的队列（如先声明排他队列，再声明非排他的同名队列）
• 消费者抛出未捕获的异常：默认情况下，handleDelivery方法中抛出未处理的异常会导致RabbitMQ关闭当前通道
• 调用channel.abort()（强制关闭通道，区别于channel.close()的主动优雅关闭）

二、排查你的监听器未触发的原因 #

1. 确认监听器注册对象：如果监听通道关闭，必须调用channel.addShutdownListener(yourListener)；监听连接关闭则调用connection.addShutdownListener(yourListener)——注册对象错误会导致事件无法触发。
2. 检查日志输出逻辑：当你调用channel.close()时，属于应用主动发起的软错误，会进入else分支，但你的代码仅打印Channel error details : {}，未输出具体错误原因，可能误以为监听器未触发。建议修改日志为：

   log.error("Channel error details : {}, reason: {}", ch, cause.getReason());
   

3. unacked消息超时场景不触发ShutdownListener：该场景下RabbitMQ仅会将消息重新入队、取消消费者，不会关闭通道/连接，因此不会触发ShutdownListener。需改用ConsumerShutdownListener监听，示例代码：

   DefaultConsumer consumer = new DefaultConsumer(channel) {
       @Override
       public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
           // 模拟处理超时不确认消息
           try {
               Thread.sleep(30 * 60 * 1000);
           } catch (InterruptedException e) {
               Thread.currentThread().interrupt();
           }
           // 不调用basicAck
       }
   
       @Override
       public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
           log.error("Consumer {} was shut down due to unacked timeout: {}", consumerTag, sig.getMessage());
       }
   };
   channel.basicConsume("your-queue", false, consumer);
   

三、测试ShutdownListener的示例代码 #

触发通道级关闭的测试代码 #

// 注册监听器到通道
channel.addShutdownListener(cause -> {
    if (cause.isHardError()) {
        log.error("Connection error with cause : {}", cause);
        Connection conn = (Connection) cause.getReference();
        if (!cause.isInitiatedByApplication()) {
            Method reason = cause.getReason();
            log.error("Rabbit Mq Connection Shutdown : {} {}", reason, cause);
        }
    } else {
        Channel ch = (Channel) cause.getReference();
        log.error("Channel error details : {}, reason: {}", ch, cause.getReason());
    }
});

// 触发通道关闭：声明同名但属性冲突的队列
channel.queueDeclare("test-queue", false, true, false, null);
// 再次声明同一队列，但排他属性改为false，触发通道错误
channel.queueDeclare("test-queue", false, false, false, null);

触发连接级关闭的测试代码 #

// 注册监听器到连接
connection.addShutdownListener(cause -> {
    log.error("Connection shut down: {}", cause.getMessage());
    if (!cause.isInitiatedByApplication()) {
        log.error("Shutdown reason: {}", cause.getReason());
    }
});

// 手动停止RabbitMQ服务，或阻断网络，即可触发监听器

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