请教：基于Thread+RabbitMQ的Java消费者实现是否正确？

Hey Manish，很高兴能帮你梳理这段RabbitMQ消费者代码！先给你一个直接的结论：你的代码可以正常运行，但存在资源浪费、线程安全隐患和资源泄漏的问题，还有不少可以优化的地方。下面咱们一步步拆解并给出改进方案：

现有实现的核心问题 #

• 不必要的Connection重复创建：你每个消费者线程都新建了一个Connection，而RabbitMQ的Connection是比较重的资源（涉及TCP连接、认证等开销），其实完全可以多个Channel共享同一个Connection（Channel是非线程安全的，每个线程用独立Channel是正确的）。
• 线程安全风险：你的count变量是普通int类型，而handleDelivery回调是由RabbitMQ客户端的IO线程触发执行的，不是你创建的消费者线程，直接修改普通int会存在可见性问题，极端情况下可能出现计数不准确。
• 资源泄漏：代码中没有任何关闭Connection/Channel的逻辑，程序退出时这些资源无法被正确释放，可能导致RabbitMQ端残留无效连接。
• 硬编码与异常处理不够健壮：队列名、交换机类型用字符串硬编码，容易出错；构造函数直接抛出检查型异常，会增加调用方的处理负担。

改进后的代码示例 #

1. 主类：共享Connection，添加资源清理钩子 #

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class ReceiveLogsDirect {
    private static final String EXCHANGE_NAME = "fanout_logs";
    private static final String QUEUE_NAME = "test";

    public static void main(String[] argv) throws Exception {
        // 初始化共享的ConnectionFactory
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.2.4");
        factory.setUsername("manish");
        factory.setPassword("mm@1234");
        
        // 创建单个共享Connection
        Connection connection = factory.newConnection();

        // 添加程序退出时的资源清理钩子，确保Connection被正确关闭
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            try {
                if (connection != null && connection.isOpen()) {
                    connection.close();
                    System.out.println("Shared connection closed successfully");
                }
            } catch (Exception e) {
                System.err.println("Failed to close connection: " + e.getMessage());
            }
        }));

        // 启动三个消费者线程，共享同一个Connection
        new Thread(new TileJobs("manish", connection)).start();
        new Thread(new TileJobs("manish1", connection)).start();
        new Thread(new TileJobs("manish2", connection)).start();
    }
}

2. 消费者类：线程安全计数，完善异常与资源处理 #

import com.rabbitmq.client.*;
import java.io.IOException;
import java.util.concurrent.atomic.AtomicInteger;

public class TileJobs implements Runnable {
    private static final String EXCHANGE_NAME = "fanout_logs";
    private static final String QUEUE_NAME = "test";
    
    private final Connection connection;
    private Channel channel;
    private final String name;
    // 使用AtomicInteger保证计数的线程安全与可见性
    private final AtomicInteger count = new AtomicInteger(0);

    public TileJobs(String name, Connection connection) {
        this.name = name;
        this.connection = connection;
        try {
            // 每个线程创建独立的Channel（Channel非线程安全，必须独占）
            this.channel = connection.createChannel();
            
            // 声明交换机与队列（幂等操作，多次调用不会重复创建）
            channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
            
            // 设置QoS，每次只接收一条未确认的消息，保证消费能力匹配
            channel.basicQos(1);
        } catch (IOException e) {
            // 把检查型异常包装为运行时异常，简化调用方处理
            throw new RuntimeException("Failed to initialize consumer channel for " + name, e);
        }
    }

    @Override
    public void run() {
        System.out.printf("✅ Consumer [%s] started, waiting for messages...%n", name);
        
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, 
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                // 原子性递增计数
                int currentCount = count.incrementAndGet();
                
                // 格式化日志，提升可读性
                System.out.printf("[%s] 已消费第 %d 条消息%n", name, currentCount);
                System.out.printf("[%s] 路由键: '%s' | 消息内容: '%s'%n", 
                        name, envelope.getRoutingKey(), message);
                
                // 手动确认消息，确保消息被处理完成后才从队列移除
                channel.basicAck(envelope.getDeliveryTag(), false);
            }

            @Override
            public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
                System.out.printf("⚠️ Consumer [%s] received shutdown signal: %s%n", name, sig.getMessage());
                // 关闭当前线程的Channel
                try {
                    if (channel != null && channel.isOpen()) {
                        channel.close();
                    }
                } catch (IOException | TimeoutException e) {
                    System.err.printf("❌ Failed to close channel for [%s]: %s%n", name, e.getMessage());
                }
            }
        };

        try {
            // 启动消费，关闭自动确认（配合basicAck手动确认）
            channel.basicConsume(QUEUE_NAME, false, consumer);
        } catch (IOException e) {
            System.err.printf("❌ Failed to start consuming for [%s]: %s%n", name, e.getMessage());
        }
    }
}

额外说明：根据需求调整队列策略 #

如果你的需求是三个消费者竞争消费同一个队列的消息（即每条消息只会被一个消费者处理，实现负载均衡），上面的代码完全适用；如果你的需求是广播消息到所有消费者（每条消息被三个消费者都处理），那需要修改队列声明逻辑，让每个消费者使用独立的临时队列：

// 替换原有的queueDeclare和queueBind代码
String uniqueQueueName = channel.queueDeclare().getQueue(); // 让RabbitMQ自动生成唯一临时队列
channel.queueBind(uniqueQueueName, EXCHANGE_NAME, "");

这样每个消费者会绑定到自己的临时队列，fanout交换机会把消息发送到所有绑定的队列，实现广播效果。

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