<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 = "/";
    
    // 工作隊(duì)列名稱(chēng)
    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. 創(chuàng)建連接工廠
        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. 創(chuàng)建連接
        Connection connection = factory.newConnection();
        
        // 3. 創(chuàng)建通道
        Channel channel = connection.createChannel();
        
        // 4. 聲明隊(duì)列（參數(shù)：隊(duì)列名稱(chēng)、是否持久化、是否排他、是否自動(dòng)刪除、額外參數(shù)）
        channel.queueDeclare(RabbitMQConstants.WORK_QUEUE_NAME, false, false, false, null);
        
        // 5. 發(fā)送10條消息
        for (int i = 0; i < 10; i++) {
            String message = "hello work queue......" + i;
            // 發(fā)送消息（參數(shù)：交換機(jī)名稱(chēng)、隊(duì)列名稱(chēng)、消息屬性、消息體）
            channel.basicPublish("", RabbitMQConstants.WORK_QUEUE_NAME, null, message.getBytes());
            System.out.println("生產(chǎn)者發(fā)送消息：" + message);
        }
        
        // 6. 關(guān)閉資源
        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. 創(chuàng)建連接工廠（同生產(chǎn)者配置）
        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. 創(chuàng)建連接
        Connection connection = factory.newConnection();
        
        // 3. 創(chuàng)建通道
        Channel channel = connection.createChannel();
        
        // 4. 聲明隊(duì)列（需與生產(chǎn)者隊(duì)列名稱(chēng)一致）
        channel.queueDeclare(RabbitMQConstants.WORK_QUEUE_NAME, false, false, false, null);
        
        // 5. 定義消息消費(fèi)回調(diào)
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody());
            System.out.println("消費(fèi)者1接收到消息：" + message);
            // 模擬任務(wù)處理耗時(shí)（100ms）
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            // 手動(dòng)確認(rèn)消息已處理（參數(shù)：消息標(biāo)識(shí)、是否批量確認(rèn)）
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };
        
        // 6. 取消消費(fèi)回調(diào)（可選）
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("消費(fèi)者1取消消費(fèi)");
        };
        
        // 7. 消費(fèi)消息（參數(shù)：隊(duì)列名稱(chēng)、是否自動(dòng)確認(rèn)、消息接收回調(diào)、取消消費(fèi)回調(diào)）
        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 {
        // 連接配置與消費(fèi)者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);
        
        // 消息消費(fèi)回調(diào)（處理耗時(shí)模擬為200ms，與消費(fèi)者1形成差異）
        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody());
            System.out.println("消費(fèi)者2接收到消息：" + message);
            try {
                Thread.sleep(200); // 處理耗時(shí)更長(zhǎng)
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        };
        
        CancelCallback cancelCallback = consumerTag -> {
            System.out.println("消費(fèi)者2取消消費(fèi)");
        };
        
        channel.basicConsume(RabbitMQConstants.WORK_QUEUE_NAME, false, deliverCallback, cancelCallback);
    }
}

// 設(shè)置每次最多接收1條未確認(rèn)消息（公平分發(fā)關(guān)鍵配置）
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());