RabbitMQ 入门教程之二

发表于 阅读次数：603 评论数： 本文字数： 3.7k 阅读时长 ≈ 3 分钟

大纲

前言

学习资源

版本说明

本文所有案例代码使用的各软件版本如下表所示：

组件版本说明
RabbitMQ Server3.8.26
RabbitMQ Client5.10.0
Erlang24.2
Java11

RabbitMQ 入门案例

简单队列模式

概念介绍

  • 这种模式只有一个生产者、一个队列、一个消费者，也被称为 “简单队列模式”。
  • 消息产生者将消息放入队列，消息的消费者监听消息队列，如果队列中有消息，就消费掉，消息被消费后，会自动从队列中删除。
  • 存在隐患，比如：消息可能没有被消费者正确处理，但在队列中已经被删除了，造成消息的丢失。
  • 应用场景：聊天（中间需要有一个过度的服务器 - P 端 与 C 端）。

案例代码

代码下载

本节所需的完整案例代码，可以直接从 GitHub 下载对应章节 rabbitmq-lesson-01。

  • Maven 依赖
1
2
3
4
5
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.10.0</version>
</dependency>
  • 生产者的代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageProperties;
import java.nio.charset.StandardCharsets;

public class MQProducer {

    // 队列名称
    public static final String QUEUE_NAME = "test";

    public static void main(String[] args) throws Exception {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.2.127");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");

        // 使用 try-with-resources 自动关闭连接和通道，确保资源释放
        try (
            // 创建连接
            Connection connection = factory.newConnection();
            // 创建信道
            Channel channel = connection.createChannel()
        ) {
            // 声明队列
            // 参数说明：
            // queue – 队列的名称
            // durable – 如果需要声明一个持久队列，则为 true（队列将在服务器重启后继续存在）
            // exclusive – 如果需要声明一个独占队列（仅限于此连接使用，连接关闭后队列自动删除），则为 true。
            // autoDelete – 如果需要声明 autoDelete 队列，则为 true（服务器将在最后一个消费者断开连接以后，自动删除该队列）
            // arguments – 队列的其他属性（构造参数）
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);

            // 发送消息
            // 参数说明：
            // exchange – 要将消息发布到的交换机，空字符串表示默认交换机
            // routingKey – 路由 Key
            // props – 消息的其他属性，比如：使用 MessageProperties.PERSISTENT_TEXT_PLAIN 属性确保消息持久化，配合持久化队列可避免服务器重启导致消息丢失
            // body – 消息内容
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, "hello".getBytes(StandardCharsets.UTF_8));
        }
    }

}

独占队列的使用说明

  • 当队列被声明为 exclusive 时，该队列仅对首次声明它的连接可见，其他连接（无论是同一客户端还是其他客户端）都无法访问该队列，并且当连接关闭时队列会被自动删除（即使设置了 durable=true）。
  • 这对于临时队列很有用，比如用于临时性任务（如 RPC 响应队列）。但是，如果是长期使用的队列，设置为 exclusive 会导致其他客户端无法访问。
  • 如果队列被声明为 exclusive，当其他连接尝试声明同名队列，并向其发送 / 消费消息，RabbitMQ 会返回错误。
  • 消费者的代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;

public class MQConsumer {

    // 队列名称
    public static final String QUEUE_NAME = "test";

    public static void main(String[] args) throws Exception {
        // 创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.2.127");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("admin");

        // 创建连接
        Connection connection = factory.newConnection();
        // 创建信道
        Channel channel = connection.createChannel();

        // 声明队列
        // 参数说明：
        // queue – 队列的名称
        // durable – 如果需要声明一个持久队列，则为 true（队列将在服务器重启后继续存在）
        // exclusive – 如果需要声明一个独占队列（仅限于此连接使用，连接关闭后队列自动删除），则为 true。
        // autoDelete – 如果需要声明 autoDelete 队列，则为 true（服务器将在最后一个消费者断开连接以后，自动删除该队列）
        // arguments – 队列的其他属性（构造参数）
        // 特别注意：如果确定队列已存在，消费者可以不声明队列。但是，强烈建议无论生产者还是消费者，都应该声明队列，确保参数可控
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);

        // 消费消息时的回调
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            String msg = new String(message.getBody(), StandardCharsets.UTF_8);
            System.out.println("Successed to consume message : " + msg);
        };

        // 取消消费时的回调（比如，在消费的时候队列已被删除掉）
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("Failed to consume message : " + consumerTag);
        };

        // 消费消息
        // 参数说明：
        // queue – 队列的名称
        // autoAck – 如果需要服务器在消息投递后自动确认消息，则为 true；如果需要客户端手动确认消息，则为 false
        // deliverCallback – 消费消息时的回调
        // cancelCallback – 取消消费时的回调
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);

        // 让消费者持续运行
        System.out.println("按回车键退出程序：");
        new Scanner(System.in).nextLine();

        // 关闭连接
        channel.close();
        connection.close();
    }

}

代码测试

  • (1) 分别启动消费者和生产者应用

  • (2) 在消费者的控制台中，会输出以下内容：

1
hello

工作队列模式

概念介绍

  • 这种模式（竞争资源 - 在工作进程之间分发任务）有一个生产者、一个队列、多个消费者，同一条消息只会被一个消费者成功消费。
  • 消息生产者将消息放入队列，消费者可以有多个。比如：消费者 1 与 消费者 2 同时监听同一个队列，消息被消费时，两个消费者共同争抢当前的消息队列内容，谁先抢到谁负责消费消息。
  • 存在隐患：高并发情况下，可能会发生某一个消息被多个消费者共同消费。
  • 应用场景：抢红包、大型项目中的资源调度（任务分配系统不需知道哪一个任务执行系统处于空闲状态，直接将任务放进到消息队列中，空闲的任务执行系统自动争抢任务）。

提示

工作队列（又称任务队列）的主要思想是避免资源密集型任务立即执行后，而不得不等待它完成的场景。相反，我们安排任务在之后执行，可以将任务封装为消息并将其发送到队列，在后台运行的工作线程将获取任务并最终执行作业。当有多个工作线程时，这些工作线程将一起处理这些任务（如下图所示），类似并行计算。

案例代码

本节将启动两个接收消息的线程（消费者）和一个发送消息的线程（生产者），以此验证两个工作线程是如何工作的（默认是由 RabbitMQ 轮询分发消息）。值得一提的是，下述代码本质上跟上面的 简单队列模式的案例代码 并没有根本区别，只是增加了一个消费者而已。

代码下载

本节所需的完整案例代码，可以直接从 GitHub 下载对应章节 rabbitmq-lesson-02。

  • Maven 依赖
1
2
3
4
5
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.10.0</version>
</dependency>
  • 工具类
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class RabbitMQUtils {

    public static ConnectionFactory connectionFactory;

    static {
        // 创建连接工厂
        connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.2.127");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
    }

    /**
     * 创建信道
     */
    public static Channel createChannel() throws Exception {
        // 创建连接
        Connection connection = connectionFactory.newConnection();

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

        return channel;
    }

}
  • 生产者的代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
import com.clay.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;

public class MQProducer {

    // 队列名称
    public static final String QUEUE_NAME = "test";

    public static void main(String[] args) throws Exception {
        // 创建信道
        Channel channel = RabbitMQUtils.createChannel();

        // 声明队列
        // 参数说明：
        // queue – 队列的名称
        // durable – 如果需要声明一个持久队列，则为 true（队列将在服务器重启后继续存在）
        // exclusive – 如果需要声明一个独占队列（仅限于此连接使用，连接关闭后队列自动删除），则为 true。
        // autoDelete – 如果需要声明 autoDelete 队列，则为 true（服务器将在最后一个消费者断开连接以后，自动删除该队列）
        // arguments – 队列的其他属性（构造参数）
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);

        // 从控制台获取需要发送的消息
        Scanner scanner = new Scanner(System.in);
        // 循环发送消息
        while (scanner.hasNext()) {
            // 消息内容
            String message = scanner.nextLine();
            System.out.println("发送消息：" + message);

            // 发送消息
            // 参数说明：
            // exchange – 要将消息发布到的交换机，空字符串表示默认交换机
            // routingKey – 路由 Key
            // props – 消息的其他属性，比如：使用 MessageProperties.PERSISTENT_TEXT_PLAIN 属性确保消息持久化，配合持久化队列可避免服务器重启导致消息丢失
            // body – 消息内容
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));
        }

        // 关闭连接
        channel.close();
    }

}
  • 消费者一的代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
import com.clay.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;

public class MQConsumer01 {

    // 队列名称
    public static final String QUEUE_NAME = "test";

    public static void main(String[] args) throws Exception {
        // 创建信道
        Channel channel = RabbitMQUtils.createChannel();

        // 声明队列
        // 参数说明：
        // queue – 队列的名称
        // durable – 如果需要声明一个持久队列，则为 true（队列将在服务器重启后继续存在）
        // exclusive – 如果需要声明一个独占队列（仅限于此连接使用，连接关闭后队列自动删除），则为 true。
        // autoDelete – 如果需要声明 autoDelete 队列，则为 true（服务器将在最后一个消费者断开连接以后，自动删除该队列）
        // arguments – 队列的其他属性（构造参数）
        // 特别注意：如果确定队列已存在，消费者可以不声明队列。但是，强烈建议无论生产者还是消费者，都应该声明队列，确保参数可控
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);

        // 消费消息时的回调
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            String msg = new String(message.getBody(), StandardCharsets.UTF_8);
            System.out.println("Successed to consume message : " + msg);
        };

        // 取消消费时的回调（比如，在消费的时候队列已被删除掉）
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("Failed to consume message : " + consumerTag);
        };

        System.out.println("消费者一等待接收消息...");

        // 消费消息
        // 参数说明：
        // queue – 队列的名称
        // autoAck – 如果需要服务器在消息投递后自动确认消息，则为 true；如果需要客户端手动确认消息，则为 false
        // deliverCallback – 消费消息时的回调
        // cancelCallback – 取消消费时的回调
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);

        // 让消费者持续运行
        System.out.println("按回车键退出程序：");
        new Scanner(System.in).nextLine();

        // 关闭连接
        channel.close();
    }

}
  • 消费者二的代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
import com.clay.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;

public class MQConsumer02 {

    // 队列名称
    public static final String QUEUE_NAME = "test";

    public static void main(String[] args) throws Exception {
        // 创建信道
        Channel channel = RabbitMQUtils.createChannel();

        // 声明队列
        // 参数说明：
        // queue – 队列的名称
        // durable – 如果需要声明一个持久队列，则为 true（队列将在服务器重启后继续存在）
        // exclusive – 如果需要声明一个独占队列（仅限于此连接使用，连接关闭后队列自动删除），则为 true。
        // autoDelete – 如果需要声明 autoDelete 队列，则为 true（服务器将在最后一个消费者断开连接以后，自动删除该队列）
        // arguments – 队列的其他属性（构造参数）
        // 特别注意：如果确定队列已存在，消费者可以不声明队列。但是，强烈建议无论生产者还是消费者，都应该声明队列，确保参数可控
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);

        // 消费消息时的回调
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            String msg = new String(message.getBody(), StandardCharsets.UTF_8);
            System.out.println("Successed to consume message : " + msg);
        };

        // 取消消费时的回调（比如，在消费的时候队列已被删除掉）
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("Failed to consume message : " + consumerTag);
        };

        System.out.println("消费者二等待接收消息...");

        // 消费消息
        // 参数说明：
        // queue – 队列的名称
        // autoAck – 如果需要服务器在消息投递后自动确认消息，则为 true；如果需要客户端手动确认消息，则为 false
        // deliverCallback – 消费消息时的回调
        // cancelCallback – 取消消费时的回调
        channel.basicConsume(QUEUE_NAME, true, deliverCallback, cancelCallback);

        // 让消费者持续运行
        System.out.println("按回车键退出程序：");
        new Scanner(System.in).nextLine();

        // 关闭连接
        channel.close();
    }

}

代码测试

  • (1) 分别启动生产者和两个消费者应用

  • (2) 在生产者的控制台中，依次输入以下内容：

1
2
3
4
5
6
7
8
AA
发送消息：AA
BB
发送消息：BB
CC
发送消息：CC
DD
发送消息：DD
  • (3) 在消费者一的控制台中，会输出以下内容
1
2
3
4
消费者一等待接收消息...
按回车键退出程序：
Successed to consume message : AA
Successed to consume message : CC
  • (4) 在消费者二的控制台中，会输出以下内容
1
2
3
4
消费者二等待接收消息...
按回车键退出程序：
Successed to consume message : BB
Successed to consume message : DD
  • (5) 观察上面两个消费者的控制台输出结果，可以发现消息默认是由 RabbitMQ 轮询分发的，而且同一条消息只会被一个消费者消费到，不会出现重复消费的情况

RabbitMQ 消费确认

RabbitMQ 支持两种消费确认机制，包括自动确认机制（默认）和手动确认机制。

自动确认机制

概念介绍

  • 自动确认机制（AutoAck）就是消息投递给消费者后立即被认为已经投递（处理）成功，这种模式需要在高吞吐量和数据传输安全性方面做权衡。因为在这种模式下，如果消息在消费者接收到之后，消费者突然宕机了，没来得及处理消息，这就会造成消息丢失。特别注意，RabbitMQ 默认使用了自动确认机制，不需要手动开启。
  • 另一方面，在这种模式下消费者那边可以接收过量的消息，没有对投递的消息数量进行限制，这样有可能使得消费者这边由于接收太多还来不及处理的消息，导致这些消息的积压，使得内存耗尽，最终这些消费者线程被操作系统杀死，所以这种模式仅适用在消费者可以高效并以某种速率处理这些消息的情况下使用。

手动确认机制

概念介绍

  • 消费者完成一个任务可能需要一段时间，如果其中一个消费者处理一个耗时长的任务并仅只完成了部分工作突然它宕机了，那么会发生什么情况呢？
  • 由于 RabbitMQ 有自动确认机制（AutoAck），一旦向消费者投递了一条消息，就会立即将该消息标记为删除。在这种情况下，突然有个消费者宕机了，将会丢失正该消费者正在处理的消息。
  • 为了保证消息在消费过程中不丢失，Rabbitmq 引入消息的手动确认机制，即消费者在接收到消息并且处理完该消息之后，主动告诉 RabbitMQ 该消息已经处理了，RabbitMQ 可以把该消息删除了。
  • 值得一提的是，手动确认的好处是可以批量确认，并且可以减少网络拥堵；但是批量确认使用得较少，为了数据传输安全性建议选择单个确认。
API 使用说明

  • Channel.basicAck(long deliveryTag, boolean multiple)

    • 用于肯定确认消息，通知 RabbitMQ 该消息成功被处理了，可以将其删除了

  • Channel.basicNack(long deliveryTag, boolean multiple, boolean requeue)

    • 用于否定确认消息，通知 RabbitMQ 该消息处理失败了

  • Channel.basicReject(long deliveryTag, boolean requeue)

    • 用于否定确认消息，通知 RabbitMQ 拒绝接收该消息，与 Channel.basicNack() 相比少了一个 multiple 参数
消息批量确认

消息批量确认使用的参数是 multiple，具体解释如下：

  • multiple = true

    • 表示批量确认 Channel 上所有未确认的消息
    • 比如：Channel 上有传送 Tag 的消息 5、6、7、8，当前 Tag 是 8，那么此时 5 ~ 8 的这些还未确认的消息都会被确认收到消费确认，如下图所示：

  • multiple = false

    • 表示单个确认消息
    • 比如：Channel 上有传送 Tag 的消息 5、6、7、8，当前 Tag 是 8，那么此时只有 8 这个消息会被确认收到消费确认，5 ~ 6 这三个消息依然不会被确认收到消费确认，如下图所示：
消息自动重新入队

消息自动重新入队使用的参数是 requeue，具体解释如下：

  • requeue = false

    • 表示丢弃 / 死信该消息

  • requeue = true

    • 表示重新将该消息重新放入队列中
    • 如果消费者由于某些原因失去连接（比如 Channel 意外关闭），导致消息未发送 ACK 确认，RabbitMQ 将了解到该消息未完全处理，并将该消息重新放入队列中。
    • 如果此时其他消费者可以处理消息，该消息将很快被重新投递给另一个消费者。这样，即使某个消费者在处理消息时突然宕机了，也可以确保不会丢失任何消息，如图所示。

案例代码

由于 RabbitMQ 默认使用的是自动确认机制（AutoAck），因此本节将介绍如何使用 RabbitMQ 的手动确认机制，并且使用的是工作队列模式。值得一提的，消费者开启手动确认机制（与生产者没有任何关系），只需要执行以下两个步骤：

  • (1) 在客户端订阅消息时，关闭自动确认机制，比如：channel.basicConsume(QUEUE_NAME, false, deliverCallback, cancelCallback);
  • (2) 在客户端消费到消息的回调（DeliverCallback）里面，手动确认消息，比如：channel.basicAck(message.getEnvelope().getDeliveryTag(), false);

代码下载

本节所需的完整案例代码，可以直接从 GitHub 下载对应章节 rabbitmq-lesson-03。

  • Maven 依赖
1
2
3
4
5
<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>5.10.0</version>
</dependency>
  • 工具类
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

public class RabbitMQUtils {

    public static ConnectionFactory connectionFactory;

    static {
        // 创建连接工厂
        connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.2.127");
        connectionFactory.setPort(5672);
        connectionFactory.setUsername("admin");
        connectionFactory.setPassword("admin");
    }

    /**
     * 创建信道
     */
    public static Channel createChannel() throws Exception {
        // 创建连接
        Connection connection = connectionFactory.newConnection();

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

        return channel;
    }

}
  • 生产者的代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
import com.clay.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.MessageProperties;

import java.nio.charset.StandardCharsets;
import java.util.Scanner;

public class MQProducer {

    // 队列名称
    public static final String QUEUE_NAME = "test";

    public static void main(String[] args) throws Exception {
        // 创建信道
        Channel channel = RabbitMQUtils.createChannel();

        // 声明队列
        // 参数说明：
        // queue – 队列的名称
        // durable – 如果需要声明一个持久队列，则为 true（队列将在服务器重启后继续存在）
        // exclusive – 如果需要声明一个独占队列（仅限于此连接使用，连接关闭后队列自动删除），则为 true。
        // autoDelete – 如果需要声明 autoDelete 队列，则为 true（服务器将在最后一个消费者断开连接以后，自动删除该队列）
        // arguments – 队列的其他属性（构造参数）
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);

        // 从控制台获取需要发送的消息
        Scanner scanner = new Scanner(System.in);
        // 循环发送消息
        while (scanner.hasNext()) {
            // 消息内容
            String message = scanner.nextLine();
            System.out.println("发送消息：" + message);

            // 发送消息
            // 参数说明：
            // exchange – 要将消息发布到的交换机，空字符串表示默认交换机
            // routingKey – 路由 Key
            // props – 消息的其他属性，比如：使用 MessageProperties.PERSISTENT_TEXT_PLAIN 属性确保消息持久化，配合持久化队列可避免服务器重启导致消息丢失
            // body – 消息内容
            channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes(StandardCharsets.UTF_8));
        }

        // 关闭连接
        channel.close();
    }

}
  • 消费者一的代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
import com.clay.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;

public class MQConsumer01 {

    // 队列名称
    public static final String QUEUE_NAME = "test";

    public static void main(String[] args) throws Exception {
        // 创建信道
        Channel channel = RabbitMQUtils.createChannel();

        // 声明队列
        // 参数说明：
        // queue – 队列的名称
        // durable – 如果需要声明一个持久队列，则为 true（队列将在服务器重启后继续存在）
        // exclusive – 如果需要声明一个独占队列（仅限于此连接使用，连接关闭后队列自动删除），则为 true。
        // autoDelete – 如果需要声明 autoDelete 队列，则为 true（服务器将在最后一个消费者断开连接以后，自动删除该队列）
        // arguments – 队列的其他属性（构造参数）
        // 特别注意：如果确定队列已存在，消费者可以不声明队列。但是，强烈建议无论生产者还是消费者，都应该声明队列，确保参数可控
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);

        // 消费消息时的回调
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            String msg = new String(message.getBody(), StandardCharsets.UTF_8);

            // 模拟消息的耗时处理
            try {
                Thread.sleep(5000);
                System.out.println("Successed to consume message : " + msg);
            } catch (Exception e) {
                e.printStackTrace();
            }

            // 手动确认消息
            // 参数说明：
            // deliveryTag – 消息的标记
            // multiple – true 表示确认所有消息，包括提供的送达标签为止的所有消息；false 仅确认提供的投放标记
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
        };

        // 取消消费时的回调（比如，在消费的时候队列已被删除掉）
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("Failed to consume message : " + consumerTag);
        };

        System.out.println("消费者一等待接收消息，处理消息较快...");

        // 关闭自动确认机制
        boolean autoAck = false;

        // 消费消息
        // 参数说明：
        // queue – 队列的名称
        // autoAck – 如果需要服务器在消息投递后自动确认消息，则为 true；如果需要客户端手动确认消息，则为 false
        // deliverCallback – 消费消息时的回调
        // cancelCallback – 取消消费时的回调
        channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, cancelCallback);

        // 让消费者持续运行
        System.out.println("按回车键退出程序：");
        new Scanner(System.in).nextLine();

        // 关闭连接
        channel.close();
    }

}
  • 消费者二的代码
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
import com.clay.rabbitmq.utils.RabbitMQUtils;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;

public class MQConsumer02 {

    // 队列名称
    public static final String QUEUE_NAME = "test";

    public static void main(String[] args) throws Exception {
        // 创建信道
        Channel channel = RabbitMQUtils.createChannel();

        // 声明队列
        // 参数说明：
        // queue – 队列的名称
        // durable – 如果需要声明一个持久队列，则为 true（队列将在服务器重启后继续存在）
        // exclusive – 如果需要声明一个独占队列（仅限于此连接使用，连接关闭后队列自动删除），则为 true。
        // autoDelete – 如果需要声明 autoDelete 队列，则为 true（服务器将在最后一个消费者断开连接以后，自动删除该队列）
        // arguments – 队列的其他属性（构造参数）
        // 特别注意：如果确定队列已存在，消费者可以不声明队列。但是，强烈建议无论生产者还是消费者，都应该声明队列，确保参数可控
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);

        // 消费消息时的回调
        DeliverCallback deliverCallback = (consumerTag, message) -> {
            String msg = new String(message.getBody(), StandardCharsets.UTF_8);

            // 模拟消息的耗时处理
            try {
                Thread.sleep(30000);
                System.out.println("Successed to consume message : " + msg);
            } catch (Exception e) {
                e.printStackTrace();
            }

            // 手动确认消息
            // 参数说明：
            // deliveryTag – 消息的标记
            // multiple – true 表示确认所有消息，包括提供的送达标签为止的所有消息；false 仅确认提供的投放标记
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
        };

        // 取消消费时的回调（比如，在消费的时候队列已被删除掉）
        CancelCallback cancelCallback = (consumerTag) -> {
            System.out.println("Failed to consume message : " + consumerTag);
        };

        System.out.println("消费者二等待接收消息，处理消息较慢...");

        // 关闭自动确认机制
        boolean autoAck = false;

        // 消费消息
        // 参数说明：
        // queue – 队列的名称
        // autoAck – 如果需要服务器在消息投递后自动确认消息，则为 true；如果需要客户端手动确认消息，则为 false
        // deliverCallback – 消费消息时的回调
        // cancelCallback – 取消消费时的回调
        channel.basicConsume(QUEUE_NAME, autoAck, deliverCallback, cancelCallback);

        // 让消费者持续运行
        System.out.println("按回车键退出程序：");
        new Scanner(System.in).nextLine();

        // 关闭连接
        channel.close();
    }

}

代码测试

  • (1) 分别启动生产者和两个消费者应用

  • (2) 在生产者的控制台中，依次输入以下内容：

1
2
3
4
AA
发送消息：AA
BB
发送消息：BB
  • (3) 由于消费者一的消息处理速度较快，在消费者一的控制台中，会输出以下内容
1
2
3
消费者一等待接收消息...
按回车键退出程序：
Successed to consume message : AA
  • (4) 在生产者发送消息 BB 后，手动将消费者二的进程杀死。按理来说 RabbitMQ 默认会轮询分发消息，即消费者二负责处理消息 BB，由于它处理消息的时间较长，在它还未处理完消息，也就是消费者二还没有执行手动 ACK 代码的时候，消费者二的进程就被杀死了。此时，RabbitMQ 控制台显示的内容如下：

  • (5) 等待一段时间后，会看到消息 BB 会被消息者一接收到了，说明消息 BB 被重新放入队列中，然后投递给能处理消息的消费者一处理了。最终，在消费者一的控制台中，会输出以下内容。这也说明使用手动确认机制后，即使消费者二在处理消息期间宕机了，也不会造成消息丢失。
1
2
3
4
消费者一等待接收消息...
按回车键退出程序：
Successed to consume message : AA
Successed to consume message : BB