RabbitMQ工作队列模式的使用解析

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.20.0</version>
</dependency>

public class RabbitMQConstants {
    // RabbitMQ连接信息
    public static final String HOST = "localhost";
    public static final int PORT = 5672;
    public static final String USERNAME = "guest";
    public static final String PASSWORD = "guest";
    public static final String VIRTUAL_HOST = "/";
    
    // 工作队列名称
    public static final String WORK_QUEUE_NAME = "work.queue";
}

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class WorkQueueProducer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(RabbitMQConstants.HOST);
        factory.setPort(RabbitMQConstants.PORT);
        factory.setUsername(RabbitMQConstants.USERNAME);
        factory.setPassword(RabbitMQConstants.PASSWORD);
        factory.setVirtualHost(RabbitMQConstants.VIRTUAL_HOST);
        
        // 2. 创建连接
        Connection connection = factory.newConnection();
        
        // 3. 创建通道
        Channel channel = connection.createChannel();
        
        // 4. 声明队列（参数：队列名称、是否持久化、是否排他、是否自动删除、额外参数）
        channel.queueDeclare(RabbitMQConstants.WORK_QUEUE_NAME, false, false, false, null);
        
        // 5. 发送10条消息
        for (int i = 0; i < 10; i++) {
            String message = "hello work queue......" + i;
            // 发送消息（参数：交换机名称、队列名称、消息属性、消息体）
            channel.basicPublish("", RabbitMQConstants.WORK_QUEUE_NAME, null, message.getBytes());
            System.out.println("生产者发送消息：" + message);
        }
        
        // 6. 关闭资源
        channel.close();
        connection.close();
    }
}

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

public class WorkQueueConsumer1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 创建连接工厂（同生产者配置）
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(RabbitMQConstants.HOST);
        factory.setPort(RabbitMQConstants.PORT);
        factory.setUsername(RabbitMQConstants.USERNAME);
        factory.setPassword(RabbitMQConstants.PASSWORD);
        factory.setVirtualHost(RabbitMQConstants.VIRTUAL_HOST);
        
        // 2. 创建连接
        Connection connection = factory.newConnection();
        
        // 3. 创建通道
        Channel channel = connection.createChannel();
        
        // 4. 声明队列（需与生产者队列名称一致）
        channel.queueDeclare(RabbitMQConstants.WORK_QUEUE_NAME, false, false, false, null);
        
        // 5. 定义消息消费回调
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody());
            System.out.println("消费者1接收到消息：" + message);
            // 模拟任务处理耗时（100ms）
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            // 手动确认消息已处理（参数：消息标识、是否批量确认）
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };
        
        // 6. 取消消费回调（可选）
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("消费者1取消消费");
        };
        
        // 7. 消费消息（参数：队列名称、是否自动确认、消息接收回调、取消消费回调）
        channel.basicConsume(RabbitMQConstants.WORK_QUEUE_NAME, false, deliverCallback, cancelCallback);
    }
}

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

public class WorkQueueConsumer2 {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 连接配置与消费者1一致
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(RabbitMQConstants.HOST);
        factory.setPort(RabbitMQConstants.PORT);
        factory.setUsername(RabbitMQConstants.USERNAME);
        factory.setPassword(RabbitMQConstants.PASSWORD);
        factory.setVirtualHost(RabbitMQConstants.VIRTUAL_HOST);
        
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        channel.queueDeclare(RabbitMQConstants.WORK_QUEUE_NAME, false, false, false, null);
        
        // 消息消费回调（处理耗时模拟为200ms，与消费者1形成差异）
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody());
            System.out.println("消费者2接收到消息：" + message);
            try {
                Thread.sleep(200); // 处理耗时更长
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };
        
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("消费者2取消消费");
        };
        
        channel.basicConsume(RabbitMQConstants.WORK_QUEUE_NAME, false, deliverCallback, cancelCallback);
    }
}

// 设置每次最多接收1条未确认消息（公平分发关键配置）
channel.basicQos(1);

channel.queueDeclare(RabbitMQConstants.WORK_QUEUE_NAME, true, false, false, null);

AMQP.BasicProperties properties = new AMQP.BasicProperties().builder()
        .deliveryMode(2) // 2表示持久化消息
        .build();
channel.basicPublish("", RabbitMQConstants.WORK_QUEUE_NAME, properties, message.getBytes());