深入浅出 RabbitMQ：工作队列实战（轮训策略VS公平策略）

public class Send {
    private final static String QUEUE_NAME = "work_mq_rr";

    public static void main(String[] argv) throws Exception {
        // 1. 配置连接工厂（同简单队列，确保虚拟主机一致）
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.229.128");
        factory.setUsername("admin");
        factory.setPassword("password");
        factory.setVirtualHost("/dev"); // 生产者与消费者需使用同一虚拟主机
        factory.setPort(5672);

        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {

            // 2. 声明队列（非持久化，非独占，不自动删除）
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);

            // 3. 循环发送10条消息
            for (int i = 0; i < 10; i++) {
                String message = "Hello World! " + i;
                // 发送消息（使用默认交换机，路由键=队列名）
                channel.basicPublish("", QUEUE_NAME, null, 
                                   message.getBytes(StandardCharsets.UTF_8));
                System.out.println(" [x] 发送消息: '" + message + "'");
            }
        }
    }
}

// 消费者1
public class Recv1 {
    private final static String QUEUE_NAME = "work_mq_rr";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.229.128");
        factory.setUsername("admin");
        factory.setPassword("password");
        factory.setVirtualHost("/dev"); // 注意：与生产者保持一致（原代码误写为/xdclass1，已修正）
        factory.setPort(5672);

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        System.out.println(" [*] Recv1 等待消息...（按CTRL+C退出）");

        // 消息处理回调（模拟处理耗时：3秒）
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            try {
                TimeUnit.SECONDS.sleep(3); // 模拟业务处理时间
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("Recv1 接收: '" + message + "'");

            // 手动确认消息（必须，否则消息会一直存在队列中）
            // 参数2：false表示仅确认当前消息，true表示确认所有小于当前tag的消息
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };

        // 关闭自动确认（autoAck=false），开启手动确认
        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
    }
}

// 消费者2（代码与Recv1一致，仅打印标识不同）
public class Recv2 {
    private final static String QUEUE_NAME = "work_mq_rr";

    public static void main(String[] argv) throws Exception {
        // 连接配置与Recv1相同
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.229.128");
        factory.setUsername("admin");
        factory.setPassword("password");
        factory.setVirtualHost("/dev");
        factory.setPort(5672);

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        System.out.println(" [*] Recv2 等待消息...（按CTRL+C退出）");

        // 消息处理回调（同样模拟3秒耗时）
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("Recv2 接收: '" + message + "'");

            // 手动确认消息
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };

        channel.basicConsume(QUEUE_NAME, false, deliverCallback, consumerTag -> {});
    }
}

// 消费者1（Recv1）添加公平策略配置
channel.queueDeclare(QUEUE_NAME, false, false, false, null);

// 关键：设置预取数为1，开启公平策略
channel.basicQos(1); // 每次只接收1条未确认的消息

// 后续的消息处理逻辑不变...

// 消费者2（Recv2）同样添加：
channel.basicQos(1);

// Recv1的处理耗时改为1秒
TimeUnit.SECONDS.sleep(1);