RabbitMQ 核心工作模式详解与 Java 实战

1.2 消费者实现

为了模拟竞争消费，我们启动两个消费者实例。关键点在于 basicConsume 的第二个参数 autoAck。如果设为 true，消息一旦发送给消费者即被确认；若为 false，则需手动 ACK。本例中为了演示方便暂时使用自动确认，实际生产中建议根据业务可靠性需求调整。

package org.example.rabbitmq.workqueues;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer1 {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 建立连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("101.43.47.137");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("study");
        connectionFactory.setPassword("study");
        connectionFactory.setVirtualHost("study");
        Connection connection = connectionFactory.newConnection();

        // 创建信道
        Channel channel = connection.createChannel();

        // 声明队列
        channel.queueDeclare("workQueues", false, false, true, null);

        // 消费消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                System.out.println("Consumer1 接收到消息：" + new String(body));
            }
        };
        
        // autoAck=true 表示自动确认
        channel.basicConsume("workQueues", true, consumer);
        
        // 保持运行以便接收消息
        Thread.sleep(100);
        
        // 资源释放
        // channel.close();
        // connection.close();
    }
}

二、Publish/Subscribe（发布/订阅）

发布/订阅模式引入了 Exchange（交换机）的概念。生产者不再直接将消息发给队列，而是发给交换机，交换机根据类型将消息路由到绑定的队列。

常见的交换机类型有三种：

1. Fanout：广播模式，忽略 RoutingKey，将消息分发给所有绑定队列。
2. Direct：定向模式，精确匹配 RoutingKey。
3. Topic：通配符模式，支持模糊匹配。

2.1 生产者实现（Fanout）

在 Fanout 模式下，我们声明一个类型为 FANOUT 的交换机，并将多个队列绑定到它。发送消息时，RoutingKey 为空即可。

package org.example.rabbitmq.fanout;

import com.rabbitmq.client.BuiltinExchangeType;
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 Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 创建连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("101.43.47.137");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("study");
        connectionFactory.setPassword("study");
        connectionFactory.setVirtualHost("study");
        Connection connection = connectionFactory.newConnection();

        // 2. 创建信道
        Channel channel = connection.createChannel();

        // 3. 声明交换机
        channel.exchangeDeclare("fanout.exchange", BuiltinExchangeType.FANOUT, false);

        // 4. 声明队列
        channel.queueDeclare("fanout.queue1", true, false, false, null);
        channel.queueDeclare("fanout.queue2", true, false, false, null);

        // 5. 交换机与队列绑定
        channel.queueBind("fanout.queue1", "fanout.exchange", "");
        channel.queueBind("fanout.queue2", "fanout.exchange", "");

        // 6. 生产消息
        for (int i = 0; i < 10; i++) {
            String msg = "hello fanout" + i;
            channel.basicPublish("fanout.exchange", "", null, msg.getBytes());
        }
        System.out.println("发送消息成功");

        // 7. 释放资源
        channel.close();
        connection.close();
    }
}

2.2 消费者实现

每个消费者监听不同的队列。由于交换机是广播模式，两个消费者都会收到相同的内容。

package org.example.rabbitmq.fanout;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("101.43.47.137");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("study");
        connectionFactory.setPassword("study");
        connectionFactory.setVirtualHost("study");
        Connection connection = connectionFactory.newConnection();

        // 创建信道
        Channel channel = connection.createChannel();

        // 声明队列
        channel.queueDeclare("fanout.queue1", true, false, false, null);

        // 消费消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                System.out.println("Consumer1 接收到消息：" + new String(body));
            }
        };
        
        channel.basicConsume("fanout.queue1", consumer);
        
        // 资源释放注释掉以保持连接
        // channel.close();
        // connection.close();
    }
}

三、Routing（路由模式）

Routing 模式使用 Direct 类型的交换机。与 Fanout 不同，它要求队列绑定交换机时必须指定 BindingKey，而生产者发送消息时需指定 RoutingKey。只有当两者完全一致时，消息才会被投递。

3.1 生产者实现

package org.example.rabbitmq.direct;

import com.rabbitmq.client.BuiltinExchangeType;
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 Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 创建连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("101.43.47.137");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("study");
        connectionFactory.setPassword("study");
        connectionFactory.setVirtualHost("study");
        Connection connection = connectionFactory.newConnection();

        // 2. 创建信道
        Channel channel = connection.createChannel();

        // 3. 声明交换机
        channel.exchangeDeclare("direct.exchange", BuiltinExchangeType.DIRECT, true);

        // 4. 声明队列
        channel.queueDeclare("direct.queue1", true, false, false, null);
        channel.queueDeclare("direct.queue2", true, false, false, null);

        // 5. 队列与交换机绑定
        channel.queueBind("direct.queue1", "direct.exchange", "a");
        channel.queueBind("direct.queue2", "direct.exchange", "a");
        channel.queueBind("direct.queue2", "direct.exchange", "b");
        channel.queueBind("direct.queue2", "direct.exchange", "c");

        // 6. 生产消息
        String msgA = "hello direct routingKey is a";
        channel.basicPublish("direct.exchange", "a", null, msgA.getBytes());
        
        String msgB = "hello direct routingKey is b";
        channel.basicPublish("direct.exchange", "b", null, msgB.getBytes());
        
        String msgC = "hello direct routingKey is c";
        channel.basicPublish("direct.exchange", "c", null, msgC.getBytes());
        
        System.out.println("发送消息成功");

        // 7. 释放资源
        channel.close();
        connection.close();
    }
}

3.2 消费者实现

package org.example.rabbitmq.direct;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("101.43.47.137");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("study");
        connectionFactory.setPassword("study");
        connectionFactory.setVirtualHost("study");
        Connection connection = connectionFactory.newConnection();

        // 创建信道
        Channel channel = connection.createChannel();

        // 声明队列
        channel.queueDeclare("direct.queue1", true, false, false, null);

        // 消费消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                System.out.println("Consumer1 接收到消息：" + new String(body));
            }
        };
        
        channel.basicConsume("direct.queue1", consumer);
        
        // 资源释放
        // channel.close();
        // connection.close();
    }
}

四、Topics（通配符模式）

Topics 模式是对 Routing 模式的扩展，允许使用通配符进行更灵活的路由匹配。

• RoutingKey 和 BindingKey 均由点号 . 分隔的单词组成。
• * 代表匹配一个单词。
• # 代表匹配零个或多个单词。

4.1 生产者实现

package org.example.rabbitmq.topics;

import com.rabbitmq.client.BuiltinExchangeType;
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 Producer {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 创建连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("101.43.47.137");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("study");
        connectionFactory.setPassword("study");
        connectionFactory.setVirtualHost("study");
        Connection connection = connectionFactory.newConnection();

        // 2. 创建信道
        Channel channel = connection.createChannel();

        // 3. 声明交换机
        channel.exchangeDeclare("topic.exchange", BuiltinExchangeType.TOPIC, true);

        // 4. 声明队列
        channel.queueDeclare("topic.queue1", true, false, false, null);
        channel.queueDeclare("topic.queue2", true, false, false, null);

        // 5. 队列与交换机绑定
        channel.queueBind("topic.queue1", "topic.exchange", "*.a.*");
        channel.queueBind("topic.queue2", "topic.exchange", "*.*.b");
        channel.queueBind("topic.queue2", "topic.exchange", "c.#");

        // 6. 生产消息
        String msgA = "hello topic routingKey is word.a.word";
        channel.basicPublish("topic.exchange", "word.a.word", null, msgA.getBytes());
        
        String msgB = "hello topic routingKey is word.word.b";
        channel.basicPublish("topic.exchange", "word.word.b", null, msgB.getBytes());
        
        String msgC = "hello topic routingKey is c.word.word.word.word.b";
        channel.basicPublish("topic.exchange", "c.word.word.word.word.b", null, msgC.getBytes());
        
        String msgD = "hello topic routingKey is c.a.b";
        channel.basicPublish("topic.exchange", "c.a.b", null, msgD.getBytes());
        
        System.out.println("发送消息成功");

        // 7. 释放资源
        channel.close();
        connection.close();
    }
}

4.2 消费者实现

package org.example.rabbitmq.topics;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer1 {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 创建连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("101.43.47.137");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("study");
        connectionFactory.setPassword("study");
        connectionFactory.setVirtualHost("study");
        Connection connection = connectionFactory.newConnection();

        // 创建信道
        Channel channel = connection.createChannel();

        // 声明队列
        channel.queueDeclare("topic.queue1", true, false, false, null);

        // 消费消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                System.out.println("Consumer1 接收到消息：" + new String(body));
            }
        };
        
        channel.basicConsume("topic.queue1", consumer);
        
        // 资源释放
        // channel.close();
        // connection.close();
    }
}

五、RPC 通信

RPC（远程过程调用）在 RabbitMQ 中通常通过两个队列配合实现：一个请求队列和一个响应队列。客户端发送请求时附带回调队列名和唯一标识（CorrelationID），服务端处理后回写到该回调队列。

5.1 客户端实现

客户端需要阻塞等待响应。这里使用了 BlockingQueue 来存储结果，并通过 correlationId 确保收到的响应属于当前请求。

package org.example.rabbitmq.rpc;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

public class Client {
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 1. 创建连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("101.43.47.137");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("study");
        connectionFactory.setPassword("study");
        connectionFactory.setVirtualHost("study");
        Connection connection = connectionFactory.newConnection();

        // 2. 创建信道
        Channel channel = connection.createChannel();

        // 3. 声明队列
        channel.queueDeclare("rpc.request.queue", true, false, false, null);
        channel.queueDeclare("rpc.response.queue", true, false, false, null);

        // 4. 发送消息
        String correlationId = UUID.randomUUID().toString();
        AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder()
                .correlationId(correlationId)
                .replyTo("rpc.response.queue")
                .build();
        String msg = "hello rpc";
        channel.basicPublish("", "rpc.request.queue", properties, msg.getBytes());

        // 5. 接收响应
        final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                if (correlationId.equals(properties.getCorrelationId())) {
                    response.offer(new String(body));
                }
            }
        };
        channel.basicConsume("rpc.response.queue", true, consumer);

        // 获取响应的结果
        String result = response.take();
        System.out.println(" [RPC_Client] Result: " + result);
    }
}

5.2 服务器实现

服务端接收请求，处理业务逻辑，并将结果发送到客户端指定的回调队列。

package org.example.rabbitmq.rpc;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Server {
    public static void main(String[] args) throws IOException, TimeoutException {
        // 1. 创建连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("101.43.47.137");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("study");
        connectionFactory.setPassword("study");
        connectionFactory.setVirtualHost("study");
        Connection connection = connectionFactory.newConnection();

        // 2. 创建信道
        Channel channel = connection.createChannel();

        // 每次接受一条消息，避免并发处理复杂化
        channel.basicQos(1);
        
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                System.out.println("Server 接收到请求：" + new String(body));

                // 响应请求
                String response = "响应";
                AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                        .correlationId(properties.getCorrelationId())
                        .replyTo("rpc.request.queue")
                        .build();
                
                channel.basicPublish("", "rpc.response.queue", props, response.getBytes());
                
                // 手动确认
                channel.basicAck(envelope.getDeliveryTag(), false);
            }
        };
        
        channel.basicConsume("rpc.request.queue", false, consumer);
    }
}

六、Publisher Confirms（发布确认）

消息丢失是分布式系统中常见的问题。RabbitMQ 提供了 Publisher Confirms 机制来解决生产者侧的消息丢失问题。主要分为三种策略：单独确认、批量确认和异步确认。

6.1 单独确认

每发送一条消息就调用 waitForConfirmsOrDie() 等待服务端确认。这种方式最安全但性能最差，因为它是同步阻塞的，且对于持久化消息需要等待磁盘写入完成。

private static void publishingMessagesIndividually() throws IOException, TimeoutException, InterruptedException {
    try (Channel channel = createChannel()) {
        // 设置信道为 confirm 模式
        channel.confirmSelect();
        channel.queueDeclare("publish.confirm.queue1", true, false, false, null);
        
        long start = System.currentTimeMillis();
        for (int i = 0; i < 200; i++) {
            String msg = "Publishing Messages Individually " + i;
            channel.basicPublish("", "publish.confirm.queue1", null, msg.getBytes());
            // 等待确认
            channel.waitForConfirmsOrDie(5000);
        }
        long end = System.currentTimeMillis();
        System.out.println("单独确认 发送 200 条消息耗时 " + (end - start));
    }
}

6.2 批量确认

为了提高性能，可以累积一定数量的消息后再等待确认。虽然减少了网络交互次数，但如果某一批次失败，重发成本较高，可能导致重复消息。

private static void publishingMessagesInBatches() throws IOException, TimeoutException, InterruptedException {
    try (Channel channel = createChannel()) {
        channel.confirmSelect();
        channel.queueDeclare("publish.confirm.queue2", true, false, false, null);
        
        long start = System.currentTimeMillis();
        int batchSize = 100;
        int flag = 0;
        
        for (int i = 0; i < 200; i++) {
            String msg = "Publishing Messages in Batches " + i;
            channel.basicPublish("", "publish.confirm.queue2", null, msg.getBytes());
            
            if (flag == batchSize) {
                channel.waitForConfirmsOrDie(5000);
                flag = 0;
            }
            flag++;
        }
        if (flag > 0) {
            channel.waitForConfirmsOrDie(5000);
        }
        
        long end = System.currentTimeMillis();
        System.out.println("批量确认 发送 200 条消息耗时 " + (end - start));
    }
}

6.3 异步确认

这是性能最高的方式。通过 addConfirmListener 注册回调接口，服务端确认后触发 handleAck 或 handleNack。我们需要维护一个已发送消息序号的集合（如 TreeSet），根据返回的 deliveryTag 更新集合状态。

private static void handlingPublisherConfirmsAsynchronously() throws IOException, TimeoutException, InterruptedException {
    try (Channel channel = createChannel()) {
        channel.confirmSelect();
        channel.queueDeclare("publish.confirm.queue3", true, false, false, null);
        
        // 有序集合，存储未 confirm 消息序号
        SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<>());
        long start = System.currentTimeMillis();
        
        // 监听
        channel.addConfirmListener(new ConfirmListener() {
            @Override
            public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                if (multiple) {
                    confirmSet.headSet(deliveryTag + 1).clear();
                } else {
                    confirmSet.remove(deliveryTag);
                }
            }

            @Override
            public void handleNack(long deliveryTag, boolean multiple) throws IOException {
                if (multiple) {
                    confirmSet.headSet(deliveryTag + 1).clear();
                } else {
                    confirmSet.remove(deliveryTag);
                }
                // 根据业务处理 nack 情况，例如重试
            }
        });

        int flag = 0;
        for (int i = 0; i < 200; i++) {
            String msg = "Handling Publisher Confirms Asynchronously " + i;
            long nextPublishSeqNo = channel.getNextPublishSeqNo();
            channel.basicPublish("", "publish.confirm.queue3", null, msg.getBytes());
            confirmSet.add(nextPublishSeqNo);
        }
        
        while (!confirmSet.isEmpty()) {
            Thread.sleep(20);
        }
        
        long end = System.currentTimeMillis();
        System.out.println("异步确认 发送 200 条消息耗时 " + (end - start));
    }
}

在实际开发中，建议优先使用异步确认模式，既能保证消息不丢失，又能获得较好的吞吐量。