Java环境下RabbitMQ直连交换机消费者无法接收消息及连接管理问题排查

嘿，我帮你捋捋这个问题的根源，你这情况我之前也碰到过！

问题出在你加的try-with-resources语句上——它会在try代码块执行完毕后，自动关闭Connection和Channel。但basicConsume是异步执行的：你调用它之后，主线程会立刻继续执行，try块结束，资源直接被关闭，这时候消费者还没来得及接收任何消息呢！

先回顾下之前的场景：没加try-with-resources时，连接一直保持打开状态，旧消费者的连接还在RabbitMQ的消费者列表里，生产者发消息的时候，RabbitMQ会把消息分发给所有绑定了对应key的消费者（这是Direct交换机的正常逻辑），所以才会出现旧消费者也收到消息的情况。

现在要解决的核心是：让消费者在真正收到并处理完消息后，再手动关闭连接，而不是依赖自动关闭。同时你用临时队列的思路是对的——每个消费者创建自己的唯一队列，绑定到Direct交换机，这样消息只会被一个消费者拿到，刚好满足「每条消息仅被一个唯一消费者接收」的需求。

给你改一下消费者的代码，用CountDownLatch来等待异步回调执行完成，确保消息处理完再关闭资源：

import java.util.concurrent.CountDownLatch;

public void consumerReceive() throws IOException, TimeoutException, InterruptedException { 
    String exchange = "directExchange";
    String key = "key1";
    ConnectionFactory factory = new ConnectionFactory(); 
    // 先创建连接和通道，不用try-with-resources，手动控制关闭时机
    Connection con = factory.newConnection();
    Channel chan = con.createChannel(); 

    chan.exchangeDeclare(exchange, "direct"); 
    // 创建临时队列，每个消费者都有自己的唯一队列
    String queueName = chan.queueDeclare().getQueue(); 
    chan.queueBind(queueName, exchange, key); 

    // 用CountDownLatch让主线程等待消息接收完成
    CountDownLatch latch = new CountDownLatch(1);

    chan.basicConsume(queueName, true, (consumerTag, msg)->{
        byte[] byteArray = msg.getBody(); 
        try { 
            Object object = deserialize(byteArray); 
            System.out.println("CONSUMER: Received message. Bye!"); 
        } catch (Exception e) {
            // 别吞异常，打印出来方便排查问题
            e.printStackTrace();
        } finally {
            // 处理完消息后，手动关闭通道和连接
            try {
                if (chan.isOpen()) chan.close();
                if (con.isOpen()) con.close();
            } catch (IOException | TimeoutException e) {
                e.printStackTrace();
            }
            // 通知主线程可以结束了
            latch.countDown();
        }
    }, consumerTag->{
        // 如果消费者被意外取消，也要关闭资源
        try {
            if (chan.isOpen()) chan.close();
            if (con.isOpen()) con.close();
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        }
        latch.countDown();
    }); 

    // 主线程等待，直到消息处理完成或消费者被取消
    latch.await();
}

关键修改点说明： #

1. 去掉try-with-resources：避免连接和通道被提前自动关闭，改为手动在回调里关闭。
2. CountDownLatch同步：因为basicConsume是异步的，主线程需要等待回调执行完毕再退出，不然刚注册完消费者程序就结束了。
3. 手动资源释放：在消息处理的finally块和消费者取消回调里都添加了资源关闭逻辑，确保无论成功失败都能释放连接。
4. 保留临时队列：每个消费者创建自己的唯一临时队列，绑定到Direct交换机的key1，这样生产者的消息只会被一个消费者接收，完美匹配你的需求。

另外还要注意一个小细节：如果生产者发送消息的时候，对应的消费者还没启动绑定队列，消息会丢失（因为临时队列是非持久化的，消息也没设置持久化）。如果需要确保消息不丢失，可以给消息和队列加上持久化配置：

• 队列持久化：chan.queueDeclare(queueName, true, false, false, null);（不过临时队列一般不需要，因为消费者断开就会被删除）
• 消息持久化：在basicPublish时设置MessageProperties.PERSISTENT_BASIC，比如chan.basicPublish(exchange, key, MessageProperties.PERSISTENT_BASIC, byteArray);

这样调整后，每个消费者接收一条消息后就会断开连接，不会再收到后续消息啦！

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