RabbitMQ 详细讲解_rabbitmq s

一、AMQP

AMQP和JMS一样，也是一个消息规范。你可能会想，已经有了JMS(Java Message Service) 。为什么还需要一个AMQP。当然是因为AMQP具备了更多优势了。

• AMQP 支持跨语言跨平台。AMQP为消息定义了线路层的协议，而JMS所定义的API的规范。这就表示，JMS的API协议能够保证所有的实现都通过通用的API来使用，但是不能保证某个JMS所发送的消息能被另外不同的JMS实现所使用。而AMQP的线路层协议规范了消息的格式，消息在生产者和消费者间传递的时候会遵循这个协议。
• AMQP 具有更加灵活和透明的消息模型。对于JMS，仅有点对点和发布-订阅两种模式消息模型可选。而AMQP通过将消息生产者和存放消息的队列解耦实现了包含但不局限于这两种的消息模型。

在JMS中，通道有助于解耦消息的生产者和消费者，但是这两者依然会和通道相耦合。生产者将消息发布到一个特定的队列或者主题中，消费者从特定的队列或主题中接收消息。通道具有了双重责任。而与之不同的是AMQP的生产者并不会直接将消息发布到队列中，AMQP在消息的生产者以及传递信息的队列之间引入了一种间接的机制：Exchange。消息的生产者将信息发布到一个Exchange上。Exchange会绑定到一个或多个队列上，他负责将信息路由到队列上。信息的消费者会从队列中提取数据并进行处理。

AMQP四种不同的Exchange

借助上面四种类型的Exchange，可以定义出不再仅限于点对点和发布-订阅的方式。

AMQP 和 JMS 的区别:

• JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式
• JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
• JMS规定了两种消息模型；而AMQP的消息模型更加丰富

二、RabbitMQ 简介

RabbitMQ 是使用Erlang编写的一个开源的消息队列，本身支持很多的协议：AMQP，XMPP, SMTP, STOMP，也正是如此，使的它变的非常重量级，更适合于企业级的开发。同时实现了一个经纪人(Broker)构架，这意味着消息在发送给客户端时先在中心队列排队。对路由(Routing)，负载均衡(Load balance)或者数据持久化都有很好的支持。
RabbitMQ是一个消息代理：它接受和转发消息。 你可以把它想象成一个邮局：当你把邮件放在邮箱里时，你可以确定邮差先生最终会把邮件发送给你的收件人。 在这个比喻中，RabbitMQ是邮政信箱，邮局和邮递员。

RabbitMQ与邮局的主要区别是它不处理纸张，而是接受，存储和转发数据消息的二进制数据块。

生产者:

(1) 生产者连接到RabbitMQ Broker，建立一个连接( Connection)开启一个信道(Channel)
(2) 生产者声明一个交换器，并设置相关属性，比如交换机类型、是否持久化等
(3) 生产者声明一个队列井设置相关属性，比如是否排他、是否持久化、是否自动删除等
(4) 生产者通过路由键将交换器和队列绑定起来
(5) 生产者发送消息至RabbitMQ Broker，其中包含路由键、交换器等信息。
(6) 相应的交换器根据接收到的路由键查找相匹配的队列。
(7) 如果找到，则将从生产者发送过来的消息存入相应的队列中。
(8) 如果没有找到，则根据生产者配置的属性选择丢弃还是回退给生产者
(9) 关闭信道。
(10) 关闭连接。

消费者:

(1) 消费者连接到RabbitMQ Broker ，建立一个连接(Connection)，开启一个信道(Channel) 。
(2) 消费者向RabbitMQ Broker 请求消费相应队列中的消息，可能会设置相应的回调函数，
(3) 等待RabbitMQ Broker 回应并投递相应队列中的消息，消费者接收消息。
(4) 消费者确认(ack) 接收到的消息。
(5) RabbitMQ 从队列中删除相应己经被确认的消息。
(6) 关闭信道。
(7)关闭连接。

三、RabbitMQ五种消息模型

RabbitMQ 提供了5种消息模式(其实是6种，只不过第六种属于RPC，所以不在此讨论)。五种消费模型分别是 基本消息模型、工作消费模型、Fanout订阅模型、Direct订阅模型、Topic订阅模型。。其中后三种都属于订阅模型。

先做些准备工作：

1 创建虚拟节点
RabbitMQ 管理平台默认地址: http://localhost:15672
默认用户名密码都是guest。

其实这一步创不创建无所谓

1. 我们使用guest 登陆后，选择admin创建一个新用户 kingfish
   
2. 创建虚拟节点/hello。我这里已经创建。
   
3. 点击创建完成的 /hello 节点，分配给 kingfish
   

2 创建RabbitMQ连接工具类

这个工具类就是和RabbitMQ建立了连接，否则后面要写很多重复代码。

/**
 * @Data: 2019/12/18
 * @Des:
 */
public class RabbitMQUtils {
    public static Connection getConection() throws IOException, TimeoutException {
        //定义连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        //设置服务地址
        factory.setHost("localhost");
        //端口
        factory.setPort(5672);
        //设置账号信息，用户名、密码、vhost
        factory.setVirtualHost("/hello");
        factory.setUsername("kingfish");
        factory.setPassword("kingfish");
        // 通过工程获取连接
        Connection connection = factory.newConnection();
        return connection;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

下面开始介绍五种消息模型:

1、基本消息模型

基本消费模型架构图:

其中 ：

• P（producer/ publisher）：生产者，一个发送消息的用户应用程序。
• C（consumer）：消费者，消费和接收有类似的意思，消费者是一个主要用来等待接收消息的用户应用程序
• 队列（红色区域）：rabbitmq内部类似于邮箱的一个概念。虽然消息流经rabbitmq和你的应用程序，但是它们只能存储在队列中。队列只受主机的内存和磁盘限制，实质上是一个大的消息缓冲区。许多生产者可以发送消息到一个队列，许多消费者可以尝试从一个队列接收数据。

生产者将消息发送到队列，消费者从队列中获取消息，队列是存储消息的缓冲区。

需要注意的是： 虽然架构图是这样画，但是本质上的信息还是通过交换机(Exchange)。在不声明交换机的情况下，使用的是RabbitMQ默认的交换机。

1、生产者代码如下：

package com.kingfish.test.rabbitmq.basic;

import com.kingfish.test.rabbitmq.RabbitMQUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

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

/**
 * @Data: 2019/12/18
 * @Des:  基本消息模型  生产者
 */
public class BasicProducer {
    private static final String QUEUE_NAME = "BaseQueueName";
    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = RabbitMQUtils.getConection();
        // 从连接中创建通道。后面大部分的操作都是通过通道完成
        Channel channel = connection.createChannel();
        // 在通道中创建一个队列
        // 【参数说明：参数一：队列名称，参数二：是否持久化；参数三：是否独占模式；参数四：消费者断开连接时是否删除队列；参数五：消息其他参数】
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 消息内容
        String message = "info : 地瓜地瓜";
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println("### 消息已发送 ： " + message);

        //关闭通道和连接
        channel.close();
        connection.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

2 . 执行生产者后，我们可以看到有一条消息已经发送到了队列中

3. 查看队列中的信息，可以看到，这个消息被发送到了默认的交换机(Exchange) 上，且RoutingKey(默认) 即队列名。

4. 消费者 代码如下

package com.kingfish.test.rabbitmq.basic;

import com.kingfish.test.rabbitmq.RabbitMQUtils;
import com.rabbitmq.client.*;

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

/**
 * @Data: 2019/12/18
 * @Des: 基本消息模型，消费者
 */
public class BasicConsumer {
    private static final String QUEUE_NAME = "BaseQueueName";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = RabbitMQUtils.getConection();
        // 从连接中创建通道。后面大部分的操作都是通过通道完成
        Channel channel = connection.createChannel();
        // 在通道中创建一个队列。如果通道中已经存在该队列，则不会重新创建。
        // 如果队列中没有该队列直接绑定会报错，所以生产者消费者中都需要声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println("地瓜消费者001 收到消息 : " + msg);
            }
        };
        // 监听队列，第二个参数：是否自动进行消息确认。
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}
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

5. 可以看到队列中消息确实被消费了。并且消费者确实接受到了消息。

2、工作消息模型

基本消费模型架构图:

在实际应用中，我们的消费者可能是集群消费。所以形成了工作消费模型。(实际上我们的生产者可能也是个集群)。

1、生产者代码如下:

这里的代码和上面的唯一的区别就是加了一个循环，发送了50次消息

public class WorkProducer {
    private static final String QUEUE_NAME = "WorkQueueName";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = RabbitMQUtils.getConection();
        // 从连接中创建通道。后面大部分的操作都是通过通道完成
        Channel channel = connection.createChannel();
        // 在通道中创建一个队列
        // 【参数说明：参数一：队列名称，参数二：是否持久化；参数三：是否独占模式；参数四：消费者断开连接时是否删除队列；参数五：消息其他参数】
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        for (int i = 0; i < 50; i++) {
            // 消息内容
            String message = "info : 地瓜地瓜 " + i;
            channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
            System.out.println("### 消息已发送 ： " + message);
        }

        //关闭通道和连接
        channel.close();
        connection.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

生产者运行后

2. 消费者1代码如下:
(消费者2 的代码和消费者1完全相同，这里就不在给出

public class WorkConsumer1 {
    private static final String QUEUE_NAME = "WorkQueueName";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = RabbitMQUtils.getConection();
        // 从连接中创建通道。后面大部分的操作都是通过通道完成
        Channel channel = connection.createChannel();
        // 在通道中创建一个队列。如果通道中已经存在该队列，则不会重新创建。
        // 如果队列中没有该队列直接绑定会报错，所以生产者消费者中都需要声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
         // 设置消费者一次只能拉取一个消息
        channel.basicQos(1);
        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println("地瓜消费者001 收到消息 : " + msg);
            }
        };
        // 监听队列，第二个参数：是否自动进行消息确认。
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}
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

注：
由于RabbitMQ的机制，会先均分给两个消费者消息。这时候如果两个消费者由于性能或其他问题处理速度不同，就会造成一个消费者早已经处理结束分配的消息，另一个消费者还在处理。这显然是不合理的。所以通过channel.basicQos(1); 设置每次值拉取一个消息。即每次只吃一个，吃完再拿，避免了上述情况。

3、订阅模型

后面的三种都属于订阅模型，只不过规则不同而已。

在之前的模式中，我们创建了一个工作队列。 工作队列背后的假设是：每个任务只被传递给一个工作人员。 在这一部分，我们将做一些完全不同的事情 - 我们将会传递一个信息给多个消费者。 这种模式被称为“发布/订阅”。

订阅模型示意图：

解读：

1、1个生产者，多个消费者
2、每一个消费者都有自己的一个队列
3、生产者没有将消息直接发送到队列，而是发送到了交换机
4、每个队列都要绑定到交换机
5、生产者发送的消息，经过交换机到达队列，实现一个消息被多个消费者获取的目的
1
2
3
4
5

X（Exchanges）：交换机一方面：接收生产者发送的消息。另一方面：知道如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。

Exchange类型有以下几种：

• Fanout：广播，将消息交给所有绑定到交换机的队列

• Direct：定向，把消息交给符合指定routing key 的队列

• Topic：通配符，把消息交给符合routing pattern（路由模式） 的队列

需要注意：Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

3.1 Fanout 订阅模型

在广播模式下，消息发送流程是这样的：

• 1） 可以有多个消费者
• 2） 每个消费者有自己的queue（队列）
• 3） 每个队列都要绑定到Exchange（交换机）
• 4） 生产者发送的消息，只能发送到交换机，交换机来决定要发给哪个队列，生产者无法决定。
• 5） 交换机把消息发送给绑定过的所有队列
• 6） 队列的消费者都能拿到消息。实现一条消息被多个消费者消费

1. 生产者代码

Fanout 模式下 : 生产者的代码中不在声明队列，直接把消息发送给交换机，交换机发送给所有绑定的队列

public class FanoutProducer {
    private static final String EXCHANGE_NAME = "fanout_exchange";
    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = RabbitMQUtils.getConection();
        // 从连接中创建通道。后面大部分的操作都是通过通道完成
        Channel channel = connection.createChannel();
        // 声明exchange，指定类型为fanout
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

        // 消息内容
        String message = "info : 地瓜地瓜 ";
        channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
        System.out.println("### 消息已发送 ： " + message);

        //关闭通道和连接
        channel.close();
        connection.close();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

2. 消费者代码
两个消费者代码相同，这里只贴出一个)

public class FanoutConsumer1 {
    private static final String EXCHANGE_NAME = "fanout_exchange";
    private static final String QUEUE_NAME = "fanout_queue_name1";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = RabbitMQUtils.getConection();
        // 从连接中创建通道。后面大部分的操作都是通过通道完成
        Channel channel = connection.createChannel();
        // 在通道中创建一个队列。如果通道中已经存在该队列，则不会重新创建。
        // 如果队列中没有该队列直接绑定会报错，所以生产者消费者中都需要声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 声明交换机，若存在则不重新声明
        channel.exchangeDeclare(EXCHANGE_NAME,   "fanout");
        // 绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
        // 设置消费者一次只能拉取一个消息
        channel.basicQos(1);
        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println("地瓜消费者001 收到消息 : " + msg);
            }
        };
        // 监听队列，第二个参数：是否自动进行消息确认。
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}
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

3.2 Direct 订阅模型

有选择性的接收消息
在订阅模式中，生产者发布消息，所有消费者都可以获取所有消息。
在路由模式中，我们将添加一个功能 - 我们将只能订阅一部分消息。 例如，我们只能将重要的错误消息引导到日志文件（以节省磁盘空间），同时仍然能够在控制台上打印所有日志消息。
但是，在某些场景下，我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。
在Direct模型下，队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由key）
消息的发送方在向Exchange发送消息时，也必须指定消息的routing key。

• P：生产者，向Exchange发送消息，发送消息时，会指定一个routing key。
• X：Exchange（交换机），接收生产者的消息，然后把消息递交给 与routing key完全匹配的队列
• C1：消费者，其所在队列指定了需要routing key 为 error 的消息
• C2：消费者，其所在队列指定了需要routing key 为 info、error、warning 的消息

1. 生产者代码

public class DirectProducer {
    private static final String EXCHANGE_NAME = "direct_exchange";
    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = RabbitMQUtils.getConection();
        // 从连接中创建通道。后面大部分的操作都是通过通道完成
        Channel channel = connection.createChannel();
        // 声明exchange，指定类型为direct
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");

        // 消息内容
        String message = "info : 地瓜地瓜 ";
        channel.basicPublish(EXCHANGE_NAME, "info", null, message.getBytes());
        System.out.println("### 消息已发送 ： " + message);

        //关闭通道和连接
        channel.close();
        connection.close();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

2. 消费者代码

public class DirectConsumer1 {
    private static final String EXCHANGE_NAME = "direct_exchange";
    private static final String QUEUE_NAME = "direct_queue_name1";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = RabbitMQUtils.getConection();
        // 从连接中创建通道。后面大部分的操作都是通过通道完成
        Channel channel = connection.createChannel();
        // 在通道中创建一个队列。如果通道中已经存在该队列，则不会重新创建。
        // 如果队列中没有该队列直接绑定会报错，所以生产者消费者中都需要声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 声明交换机，若存在则不重新声明
        channel.exchangeDeclare(EXCHANGE_NAME,   "direct");
        // 绑定队列到交换机。绑定的routingKey为 info 和error。即
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "info");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "error");
        // 设置消费者一次只能拉取一个消息
        channel.basicQos(1);
        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println("地瓜消费者001 收到消息 : " + msg);
            }
        };
        // 监听队列，第二个参数：是否自动进行消息确认。
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}
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

消费者2 与消费者1的不同之处 ： 消费者2绑定了info和error的routingKey。

上述代码执行后，消费者2可以接收到生产者的消息。而消费者1不可以。因为消费者1绑定的 routingKey 是error。而生产者发送的routeringKey 是info。

3.3 Topic 订阅模型

Topic类型的Exchange与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！

Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

#：匹配一个或多个词
*：匹配不多不少恰好1个词

如：

audit.#：能够匹配audit.irs.corporate 或者 audit.irs
audit.*：只能匹配audit.irs

1. 生产者代码：

public class TopicProducer {
    private static final String EXCHANGE_NAME = "topic_exchange";
    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = RabbitMQUtils.getConection();
        // 从连接中创建通道。后面大部分的操作都是通过通道完成
        Channel channel = connection.createChannel();
        // 声明exchange，指定类型为fanout
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");

        // 消息内容
        String message = "info : 地瓜地瓜 ";
        channel.basicPublish(EXCHANGE_NAME, "www.baidu.com", null, message.getBytes());
        System.out.println("### 消息已发送 ： " + message);

        //关闭通道和连接
        channel.close();
        connection.close();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

2. 消费者1 代码

package com.kingfish.test.rabbitmq.topic;

import com.kingfish.test.rabbitmq.RabbitMQUtils;
import com.rabbitmq.client.*;

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

/**
 * @Data: 2019/12/18
 * @Des:
 */
public class TopicConsumer1 {
    private static final String EXCHANGE_NAME = "topic_exchange";
    private static final String QUEUE_NAME = "topic_queue_name1";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = RabbitMQUtils.getConection();
        // 从连接中创建通道。后面大部分的操作都是通过通道完成
        Channel channel = connection.createChannel();
        // 在通道中创建一个队列。如果通道中已经存在该队列，则不会重新创建。
        // 如果队列中没有该队列直接绑定会报错，所以生产者消费者中都需要声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        // 声明交换机，若存在则不重新声明
        channel.exchangeDeclare(EXCHANGE_NAME,   "topic");
        // 绑定队列到交换机
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "www.#");
        // 设置消费者一次只能拉取一个消息
        channel.basicQos(1);
        // 定义队列的消费者
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            // 获取消息，并且处理，这个方法类似事件监听，如果有消息的时候，会被自动调用
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                // body 即消息体
                String msg = new String(body);
                System.out.println("地瓜消费者001 收到消息 : " + msg);
            }
        };
        // 监听队列，第二个参数：是否自动进行消息确认。
        channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}
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

消费者2和1不同的地方

上面的例子中，生产者生产消息后，消费者1和2都会收到消息。因为消费者2 routingKey 是 www.#
消费者2 routingKey 是 ww.baidu.*。结合上面说的 # 匹配若干，* 匹配唯一不难推论出来。

四、RabbitMQ 消息可靠性

更为详细的可靠性文章请移步（本章节部分参考该文）： https://mp.weixin.qq.com/s/dYH0wAYYiXuQwiBqc7wMjg

---

RabbitMQ 整个工作流程可以分为以下三步：

第一步：生产端到RabbitMQ

这一步可以通过事务机制或者Comfirm机制来保证。需要注意的是，事务机制虽然保证了消息投递端的可靠性，但因为每次投递都开启了事务，所以性能较低，一般不推荐使用，一般使用 Confirm 机制。

第二步：RabbitMQ

这一步是保证 RabbitMQ 由于某些故障消息如何保证不丢失。

• 持久化 ： 持久化可以提高 RabbitMQ 的可靠性，以免在 RabbitMQ 意外宕机时数据不会丢失，RabbitMQ 的 Exchange、Queue 以及 Message 都是支持持久化的，Exchange 和 Queue 通过在声明的时候将 durable 参数置为 true 即可实现，而消息的持久化则需要将投递模式（BasicProperties 中的 deliveryMode 属性）设置为2（PERSISTENT）。但需要注意的是，必须同时将 Queue 和 Message 持久化才能保证消息不丢失，仅设置 Queue 持久化，重启之后 Message 会丢失，反之仅设置消息的持久化，重启之后 Queue 消失，既而 Message 也丢失。
• 集群 ： 上述持久化的操作保证了消息在 RabbitMQ 宕机时不会丢失，但却不能避免单机故障且无法修复（比如磁盘损毁）而引起的消息丢失，并且在故障发生时 RabbitMQ 不可用。这时就需要引入集群，由于 RabbitMQ 是基于 Erlang 编写的，所以其天生支持分布式，而不需要像 Kafka 那样要通过 Zookeeper 来实现，RabbitMQ Cluster 集群共有两种模式

  1. 普通模式 ： 普通模式下，集群中的 RabbitMQ 会同步 Vhost、Exchange、Binding、Queue 的元数据（即其本身的数据，例如名称、属性等）以及 Message 结构，而不会同步 Message 数据，也就是说，如果集群中某台机器 RabbitMQ 宕掉了，则该节点上的 Message 不可用，直至该节点恢复。
     镜像模式

  2. 镜像队列 ：相当于配置了副本，绝大多数分布式的东西都有多副本的概念来确保 HA（High Availability）。在镜像队列中，如果主节点（master）在此特殊时间内挂掉，可以自动切换到从节点（slave），这样有效的保证了高可用性，除非整个集群都挂掉。

第三步：RabbitMQ到消费者

这一步是为了保证消息被正确的消费，因此我们可以使用手动签收的机制

为了保证消息从队列可靠地达到消费者，RabbitMQ 提供了消息确认机制（message acknowledgement）。消费者在订阅队列时，可以指定 autoAck 参数，当 autoAck 等于 fals e时，RabbitMQ 会等待消费者显式地回复确认信号后才从内存（或者磁盘）中移去消息（实质上是先打上删除标记，之后再删除）。当 autoAck 等于 true 时，RabbitMQ 会自动把发送出去的消息置为确认，然后从内存（或者磁盘）中删除，而不管消费者是否真正的消费到了这些消息。

对于 RabbitMQ 而言，队列中的消息分成了两个部分：一部分是等待投递给消费者的消息；一部分是已经投递给消费者，但是还没有收到消费者确认信号的消息。如果 RabbitMQ 一直没有收到消费者的确认信号，并且消费此消息的消费者已经断开连接，则 RabbitMQ 会安排该消息重新进入队列，等待投递给下一个消费者。RabbitMQ 判断此消息是否需要重新投递的唯一依据是消费该消息的消费者连接是否已经断开，这种设计允许消费者消费一条消息很久很久。

如果消息消费失败，也可以调用 Basic.Reject 或者 Basic.Nack 来拒绝当前消息，但需要注意的是，如果只是简单的拒绝那么消息将会丢失，需要将相应的 requeue 参数设置为 true，RabbitMQ 才会将这条消息重新存入队列。而如果 requeue 参数设置为 false 的话，RabbitMQ 立即会把消息从队列中移除，而不会把它发送给新的消费者。

basicNack 和 basicReject 作用基本相同，主要差别在于前者可以拒绝多条，后者只能拒绝单条，另外basicNack 不是 AMQP 0-9-1 标准。

// 确认消息
channel.basicAck(deliveryTag, multiple);
// 拒绝消息
channel.basicNack(deliveryTag, multiple, requeue);
// 拒绝消息
channel.basicReject(deliveryTag, requeue)
1
2
3
4
5
6

---

1. RabbitMQ 持久化

上面介绍的五种消息模型，本质上都是通过交换机绑定队列进行交互。所以当RabbitMQ宕机后消息就会丢失。所以我们需要持久化保存消息。
想要持久化消息就要先持久化队列，想要持久化队列就要先持久化交换机。

交换机的持久化 :

队列的持久化：

消息的持久化：

2. RabbitMQ 签收机制

2.1 RabbitMQ生产者签收机制

RabbitMQ中对生产者提供了两种签收机制 ： 事务机制和Confirm机制

2.1.1 事务机制

通过 AMQP 事务机制实现，这也是 AMQP 协议层面提供的解决方案

• RabbitMQ 中与事务机制有关的方法有三个：txSelect(), txCommit()以及 txRollback(),
• txSelect 用于将当前 channel 设置成 transaction 模式 txCommit 用于提交事务,
• txRollback 用于回滚事务，在通过 txSelect 开启事务之后，我们便可以发布消息给 broker 代理服务器了，如果
• txCommit 提交成功了，则消息一定到达了 broker 了 如果在 txCommit执行之前 broker
  异常崩溃或者由于其他原因抛出异常，这个时候我们便可以捕获异常通过 txRollback 回滚事务了。
  关键代码：

	channel.txSelect();
	channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
	channel.txCommit();
1
2
3

生产者： 主要部分

try {
    // 启用事务模式
    channel.txSelect();
     // 发送内容【参数说明：参数一：交换机名称；参数二：队列名称，参数三：消息的其他属性-routing headers，此属性为MessageProperties.PERSISTENT_TEXT_PLAIN用于设置纯文本消息存储到硬盘；参数四：消息主体】
    channel.basicPublish("exchangeKey_direct", "routerKey1", null, content.getBytes("UTF-8"));
     System.out.println("已发送消息：" + content);
    int i = 1/ 0;
    // 事务提交
    channel.txCommit();
} catch (Exception e) {
    // 事务回滚
    channel.txRollback();
    e.printStackTrace();
    System.out.println("发送出错，消息回滚。" + content);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

此种模式很耗时且采用这种方式 降低了 Rabbitmq 的消息吞吐量

2.2.2. Confirm模式

*通过将 channel 设置成 confirm 模式来实现。需要注意，事务方式和Confirm两种模式不能共存
生产者将信道设置成confirm模式，一旦信道进入confirm模式，所有在该信道上面发布的消息都将会被指派一个唯一的ID(从1开始)，一旦消息被投递到所有匹配的队列之后，broker就会发送一个确认给生产者(包含消息的唯一ID)，这就使得生产者知道消息已经正确到达目的队列了，如果消息和队列是可持久化的，那么确认消息会在将消息写入磁盘之后发出，broker回传给生产者的确认消息中delivery-tag域包含了确认消息的序列号，此外broker也可以设置basic.ack的multiple域，表示到这个序列号之前的所有消息都已经得到了处理；
confirm模式最大的好处在于他是异步的，一旦发布一条消息，生产者应用程序就可以在等信道返回确认的同时继续发送下一条消息，当消息最终得到确认之后，生产者应用便可以通过回调方法来处理该确认消息，如果RabbitMQ因为自身内部错误导致消息丢失，就会发送一条nack消息，生产者应用程序同样可以在回调方法中处理该nack消息；

在编程中我们可以选择下面的几种编程方式

• 普通 confirm 模式：每发送一条消息后，调用 waitForConfirms()方法，等待服务器端confirm。实际上是一种串行 confirm 了。
• 批量 confirm 模式：每发送一批消息后，调用 waitForConfirms()方法，等待服务器端confirm。
• 异步 confirm 模式：提供一个回调方法，服务端 confirm 了一条或者多条消息后 Client 端会回调这个方法。

1、普通 confirm 模式

                // 生产者通过调用confirmSelect 方法将 channel 设置为 confirm 模式
                channel.confirmSelect();
                // 发送内容【参数说明：参数一：交换机名称；参数二：队列名称，参数三：消息的其他属性-routing headers，此属性为MessageProperties.PERSISTENT_TEXT_PLAIN用于设置纯文本消息存储到硬盘；参数四：消息主体】
                channel.basicPublish("exchangeKey_direct", "routerKey1", null, content.getBytes("UTF-8"));
                // 判断消息是否发送成功
                if (channel.waitForConfirms()) {
                    System.out.println("已发送消息：" + content);
                } else {
                    System.out.println("消息发送失败，消息内容 ：" + content);
                }
1
2
3
4
5
6
7
8
9
10

2、批量 confirm 模式
批量 confirm 模式稍微复杂一点，客户端程序需要定期（每隔多少秒）或者定量（达到多少条）或者两则结合起来publish 消息，然后等待服务器端 confirm, 相比普通 confirm 模式，批量极大提升 confirm 效率，但是问题在于一旦出现 confirm 返回 false 或者超时的情况时，客户端需要将这一批次的消息全部重发，这会带来明显的重复消息数量，并且，当消息经常丢失时，批量 confirm 性能应该是不升反降的。

                // 生产者通过调用confirmSelect 方法将 channel 设置为 confirm 模式
                channel.confirmSelect();
                for (int i = 0; i < 10; i++) {
                    // 发送内容【参数说明：参数一：交换机名称；参数二：队列名称，参数三：消息的其他属性-routing headers，此属性为MessageProperties.PERSISTENT_TEXT_PLAIN用于设置纯文本消息存储到硬盘；参数四：消息主体】
                    channel.basicPublish("exchangeKey_direct", "routerKey1", null, content.getBytes("UTF-8"));
                }
                // 判断消息是否发送成功
                if (channel.waitForConfirms()) {
                    System.out.println("已发送消息：" + content);
                } else {
                    System.out.println("消息发送失败，消息内容 ：" + content);
                }
1
2
3
4
5
6
7
8
9
10
11
12

3、异步 confirm 模式
Channel 对象提供的 ConfirmListener()回调方法只包含 deliveryTag（当前 Chanel 发出的消息序号），我们需要自己为每一个 Channel 维护一个 unconfirm 的消息序号集合，每 publish 一条数据，集合中元素加 一，每回调一次 handleAck方法，unconfirm 集合删掉相应的一条（multiple=false）或多条（multiple=true）记录。从程序运行效率上看，这个unconfirm 集合最好采用有序集合 SortedSet 存储结构。
实际上，SDK 中的 waitForConfirms()方法也是通过 SortedSet维护消息序号的。

                // 生产者通过调用confirmSelect 方法将 channel 设置为 confirm 模式
                channel.confirmSelect();
                // 创建一个列表，用于维护消息发送的情况
                final SortedSet<Long> confirmSet = Collections.synchronizedSortedSet(new TreeSet<Long>());
                // 添加Confirm监听器，监听消息发送的状态。
                channel.addConfirmListener(new ConfirmListener() {
                    //每回调一次handleAck方法，confirmSet 删掉相应的一条（multiple=false）或多条（multiple=true）记录。
                    @Override
                    public void handleAck(long deliveryTag, boolean multiple) throws IOException {
                        if (multiple) {
                            System.out.println("移除集合中的多条记录--");
                            confirmSet.headSet(deliveryTag + 1).clear(); //用一个SortedSet, 返回此有序集合中小于end的所有元素。
                        } else {
                            System.out.println("--multiple false--");
                            confirmSet.remove(deliveryTag);
                        }
                    }
                    // 消息签收失败时调用
                    @Override
                    public void handleNack(long deliveryTag, boolean multiple) throws IOException {
//                        System.out.println("Nack, SeqNo: " + deliveryTag + ", multiple: " + multiple);
//                        if (multiple) {
//                            confirmSet.headSet(deliveryTag + 1).clear();
//                        } else {
//                            confirmSet.remove(deliveryTag);
//                        }
                    }
                });

                while (true) {
                    // Confirm模式下，返回下一条要发布的消息的序列号。
                    long nextSeqNo = channel.getNextPublishSeqNo();
                    Thread.sleep(2000);
                    channel.basicPublish("exchangeKey_direct", "routerKey1", null, content.getBytes("UTF-8"));
                    confirmSet.add(nextSeqNo);
                }
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

2.2. RabbitMQ消费者者的签收机制:

对于消费者的消息签收有有自动和手动两种。
自动签收 ： 并不会管消息的处理即其它问题。类似于签收快递，自动签收则是将快递放入快递柜，至于后续，不管你是否接收，以及快递是否有问题（程序执行不通过），都不会有所反应。
手动签收 ： 是由程序确认是否签收。即自己本人签收快递，即细致检查（程序处理）后，如果没有问题，则会确定签收，若有问题，可以拒签。

2.2.1. 自动签收

通过 String basicConsume(String queue, boolean autoAck, String consumerTag, Consumer callback) throws IOException; 方法或其重载方法中的 autoAck 参数设置是否自动签收。为true则自动签收

              // 创建通道
                Channel channel = conn.createChannel();
                // 创建订阅器，并接受消息。 第一个参数 : 队列名， 第二个参数签收方式 : false，设置成手动签收模式；true，则自动签收模式
                channel.basicConsume("queueName1", true, "", new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                               byte[] body) throws IOException {
                        String routingKey = envelope.getRoutingKey(); // 队列名称
                        String contentType = properties.getContentType(); // 内容类型
                        String content = new String(body, "utf-8"); // 消息正文
                        System.out.println("routingKey :" + routingKey + "   contentType : " + contentType + "   消息正文：" + content);
//                        channel.basicAck(envelope.getDeliveryTag(), false); // 手动确认消息【参数说明：参数一：该消息的index；参数二：是否批量应答，true批量确认小于index的消息】
                    }
                });
1
2
3
4
5
6
7
8
9
10
11
12
13
14

2.2.2. 手动签收

通过 String basicConsume(String queue, boolean autoAck, String consumerTag, Consumer callback) throws IOException; 方法或其重载方法中的 autoAck 参数设置为false，通过 void basicAck(long deliveryTag, boolean multiple) 方法来进行手动签收。

                // 创建通道
                Channel channel = conn.createChannel();
                // 创建订阅器，并接受消息。 第一个参数 : 队列名， 第二个参数签收方式 : false，设置成手动签收模式。true，则自动签收模式
                channel.basicConsume("queueName1", false, "", new DefaultConsumer(channel) {
                    @Override
                    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                                               byte[] body) throws IOException {
                        String routingKey = envelope.getRoutingKey(); // 队列名称
                        String contentType = properties.getContentType(); // 内容类型
                        String content = new String(body, "utf-8"); // 消息正文
                        System.out.println("routingKey :" + routingKey + "   contentType : " + contentType + "   消息正文：" + content);
//                        channel.basicAck(envelope.getDeliveryTag(), false); // 手动确认消息【参数说明：参数一：该消息的index；参数二：是否批量应答，true批量确认小于index的消息】
                    }
                });
1
2
3
4
5
6
7
8
9
10
11
12
13
14

额外还有一些方法：

//扔掉消息
channel.BasicReject(result.DeliveryTag, false);

//退回消息
channel.BasicReject(result.DeliveryTag, true);

//批量退回或删除,中间的参数 是否批量 true是/false否 (也就是只一条)
channel.BasicNack(result.DeliveryTag, true, true);

//补发消息 true退回到queue中/false只补发给当前的consumer；BasicRecover方法则是进行补发操作，其中的参数如果为true是把消息退回到queue但是有可能被其它的consumer接收到，设置为false是只补发给当前的consumer
channel.BasicRecover(true);
1
2
3
4
5
6
7
8
9
10
11

注： 生产者的签收成功指的是消息被RabbitMQ签收，并不是指被消费。
消费者的签收成功指的是RabbitMQ中的消息被消费。

五、Spring 中整合 RabbitMQ

1. 创建 rabbitmq.properties.配置文件配置基础信息

#访问RabbitMQ服务器的账户,默认是guest
rabbitmq.username=guest
#访问RabbitMQ服务器的密码,默认是guest
rabbitmq.password=guest
#RabbitMQ服务器地址，默认值"localhost
rabbitmq.host=localhost
#RabbitMQ服务端口，默认值为5672
rabbitmq.port=5672
#hannel的缓存数量，默认值为25
rabbitmq.channelCacheSize=50
#缓存连接模式，默认值为CHANNEL(单个connection连接，连接之后关闭，自动销毁)
rabbitmq.cacheMode=CHANNEL

1
2
3
4
5
6
7
8
9
10
11
12
13

2. 发送与接收消息

2.1. 使用RabbitTemplate 发送与接收消息

创建RabbitMQConfig 配置类：

package com.kingfish.common.config.mq;

import com.kingfish.pojo.handler.RabbitMQHandler;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerEndpoint;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;

import java.io.UnsupportedEncodingException;

/**
 * @Data: 2019/10/9
 * @Des:
 */
@Configuration
@PropertySource(value = {"classpath:rabbitmq.properties"})
@EnableRabbit
public class RabbitMQConfig {
    @Value("${rabbitmq.username}")
    private String username;
    @Value("${rabbitmq.password}")
    private String password;
    @Value("${rabbitmq.host}")
    private String host;
    @Value("${rabbitmq.port}")
    private int port;
    @Value("${rabbitmq.channelCacheSize}")
    private int channelCacheSize;
    @Value("${rabbitmq.cacheMode}")
    private String cacheMode;

    @Bean
    public ConnectionFactory cachingConnectionFactory() {
        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
        cachingConnectionFactory.setHost(host);
        cachingConnectionFactory.setChannelCacheSize(channelCacheSize);
        cachingConnectionFactory.setCacheMode(CachingConnectionFactory.CacheMode.CHANNEL);
        return cachingConnectionFactory;
    }

    /**
     * 该类封装了对 RabbitMQ 的管理操作
     *
     * @param connectionFactory
     * @return
     */
    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
        // 声明队列
        Queue queueName1 = new Queue("queueName1");
        rabbitAdmin.declareQueue(queueName1);
        // 声明交换机
        rabbitAdmin.declareExchange(new DirectExchange("exchangeKey_direct", true, false));
        // 使用BindingBuilder进行绑定
        rabbitAdmin.declareBinding(BindingBuilder.bind(queueName1).to(new DirectExchange("exchangeKey_direct")).with("routerKey1"));
        return rabbitAdmin;
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        // 设置消息回调
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.out.println("rabbit return success" + message.toString() + "===" + replyText + "===" + exchange + "===" + routingKey);
            }
        });
        // 消息回调结果
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
            if (!ack) {
                System.out.println("rabbit 消息发送失败" + cause + correlationData.toString());
            } else {
                System.out.println("rabbit 消息发送成功 ");
            }
        });
        return rabbitTemplate;
    }


    @Bean
    public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
         // 设置QOS,保证同一时间一个消费者只能消费一条消息
        factory.setPrefetchCount(1);
        //初始化消费者数量
        factory.setConcurrentConsumers(1);
        //最大消费者数量
        factory.setMaxConcurrentConsumers(1);

        // 设置应答模式
        factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);

        return factory;
    }

	/**
	* 可以在这个bean中设置接收消息方式
	*/
    @Bean
    public RabbitListenerConfigurer rabbitListenerConfigurer() {
        return new RabbitListenerConfigurer() {
            @Override
            public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {
                // 方式1 ： 直接处理
                SimpleRabbitListenerEndpoint endpoint = new SimpleRabbitListenerEndpoint();
                endpoint.setId("0");
                endpoint.setQueueNames("queueName1");
                endpoint.setMessageListener(message -> {
                    // 直接在这里处理消息信息
                    try {
                        System.out.println("endpoint1处理消息的逻辑 : " + new String(message.getBody(), "utf-8"));
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                    }
                });

                //方式2 使用适配器来处理消息 RabbitMQHandler 来处理信息
                SimpleRabbitListenerEndpoint endpoint2 = new SimpleRabbitListenerEndpoint();
                endpoint2.setId("1");
                endpoint2.setQueueNames("queueName1");
                System.out.println("endpoint2处理消息的逻辑");
                // 绑定pojo，并指定默认处理方法为onMessage
                endpoint2.setMessageListener(new MessageListenerAdapter(new RabbitMQHandler(), "onMessage"));

                //注册endpoint
                registrar.registerEndpoint(endpoint);
                registrar.registerEndpoint(endpoint2);
            }
        };
    }
}
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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146

public class RabbitMQHandler {
    public void onMessage(Object object) {
        System.out.println("RabbitMQHandler : " + object.toString());
    }
}
1
2
3
4
5

2.2 发送消息和接收消息的代码

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @Data: 2019/10/10
 * @Des:
 */
@RestController
@RequestMapping("mq")
public class MQController {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @RequestMapping("send")
    public String sendMessage() {
        rabbitTemplate.convertAndSend("exchangeKey_direct", "routerKey1",
                "发送的信息" + new SimpleDateFormat("yyyy-MM-dd hh:mm;ss").format(new Date()));
        return "ok";
    }

    @RequestMapping("receive")
    public String receiveMessage() throws Exception {
        // 接收消息
        Message receive = rabbitTemplate.receive("queueName1");
        System.out.println("消息正文：" + new String(receive.getBody(), "utf-8"));

        return new String(receive.getBody(), "utf-8");
    }
}
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

2.3 测试

1. 调用 http://localhost:8080/mq/send 接口，发送一个消息到rabbit
   
2. 调用 http://localhost:8080/mq/receive接口，接收消息
   

3. 使用消息驱动接收消息

使用上面的方式，很明显没有办法异步监听消息。如果我们不调用receive接口，便无法接收到rabbit中的消息。而使用消息驱动方式则可以解决这个问题。

3.1 指定消息驱动类

RabbitCMQConfig配置类

package com.kingfish.common.config.mq;

import com.kingfish.pojo.handler.RabbitMQHandler;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerEndpoint;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;
import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;

import java.io.UnsupportedEncodingException;

/**
 * @Data: 2019/10/9
 * @Des:
 */
@Configuration
@PropertySource(value= {"classpath:rabbitmq.properties"})
@EnableRabbit	
public class RabbitMQConfig {
    @Value("${rabbitmq.username}")
    private String username;
    @Value("${rabbitmq.password}")
    private String password;
    @Value("${rabbitmq.host}")
    private String host;
    @Value("${rabbitmq.port}")
    private int port;
    @Value("${rabbitmq.channelCacheSize}")
    private int channelCacheSize;
    @Value("${rabbitmq.cacheMode}")
    private String cacheMode;

    @Bean
    public ConnectionFactory cachingConnectionFactory(){
        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
        cachingConnectionFactory.setHost(host);
        cachingConnectionFactory.setChannelCacheSize(channelCacheSize);
        return cachingConnectionFactory;
    }

    @Bean
    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory){
        return new RabbitAdmin(connectionFactory);
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setExchange("exchangeKey_direct");
        return rabbitTemplate;
    }


    @Bean
    public RabbitListenerContainerFactory<?> rabbitListenerContainerFactory(ConnectionFactory connectionFactory){
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        return factory;
    }
	
    @Bean
    public RabbitListenerConfigurer rabbitListenerConfigurer(){
       return new RabbitListenerConfigurer() {
            @Override
            public void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {
                    // 方式1 ： 直接处理
                    SimpleRabbitListenerEndpoint endpoint = new SimpleRabbitListenerEndpoint();
                    endpoint.setId("0");
                    endpoint.setQueueNames("queueName1");
                    endpoint.setMessageListener(message -> {
                        // 直接在这里处理消息信息
                        try {
                            System.out.println("endpoint1处理消息的逻辑 : " +  new String(message.getBody(), "utf-8"));
                        } catch (UnsupportedEncodingException e) {
                            e.printStackTrace();
                        }
                    });

                    //方式2 使用适配器来处理消息 RabbitMQHandler 来处理信息
                    SimpleRabbitListenerEndpoint endpoint2 = new SimpleRabbitListenerEndpoint();
                    endpoint2.setId("1");
                    endpoint2.setQueueNames("queueName1");
                    System.out.println("endpoint2处理消息的逻辑");
                    // 绑定pojo，并指定默认处理方法为onMessage
                    endpoint2.setMessageListener(new MessageListenerAdapter(new RabbitMQHandler(),"onMessage"));

                    //注册endpoint
                    registrar.registerEndpoint(endpoint);
                    registrar.registerEndpoint(endpoint2);

            }
        };
    }
}
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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104

消息驱动类 RabbitMQHandler

public class RabbitMQHandler {

    public void onMessage(Object object) throws Exception {
    	// 接收到消息
        System.out.println("RabbitMQHandler : " + object.toString());
    }
}
1
2
3
4
5
6
7

当有消息发送时，RabbitMQHandler.onMessage方法来接收信息并进行需要的处理。

程序运行如下：

1. 调用 http://localhost:8080/mq/send 接口，发送一个消息到rabbit
2. 可以直接接收到消息内容。

3.3 使用注解方式配置

使用注解方式配置也很简单，声明一个如下的类即可

@Component
// 监听注解，也可以直接写在方法上
@RabbitListener(queues = "queueName1")
public class RabbitMQListener {

    @RabbitHandler()
    public void process(String hello, Channel channel, Message message) throws IOException {
        System.out.println("HelloReceiver收到 : " + hello + "收到时间" + new Date());
        try {
            //告诉服务器收到这条消息 已经被我消费了 可以在队列删掉 这样以后就不会再发了 否则消息服务器以为这条消息没处理掉 后续还会在发
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            System.out.println("receiver success");
        } catch (IOException e) {
            e.printStackTrace();
            //丢弃这条消息
            //channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);
            System.out.println("receiver fail");
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

写在方法上 ：

@Component
public class Listener {
	
    @RabbitListener(bindings = @QueueBinding(
            value = @Queue(value = "spring.test.queue", durable = "true"),
            exchange = @Exchange(
                    value = "spring.test.exchange",
                    ignoreDeclarationExceptions = "true",	// 忽略交换机声明异常，使用已有交换机
                    type = ExchangeTypes.TOPIC				// Topic 类型
            ),
            key = {"#.#"}))			// 通配符
    // 接收的消息是什么类型就用什么类型接收
    public void listen(@Payload String msg, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Channel channel) throws IOException){
        System.out.println("接收到消息：" + msg);
         // 手动应答确认消费成功
        channel.basicAck(deliveryTag,false);
    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

• @Componet：类上的注解，注册到Spring容器
• @RabbitListener：方法上的注解，声明这个方法是一个消费者方法，需要指定下面的属性：
  • bindings：指定绑定关系，可以有多个。值是@QueueBinding的数组。@QueueBinding包含下面属性：
    • value：这个消费者关联的队列。值是@Queue，代表一个队列
    • exchange：队列所绑定的交换机，值是@Exchange类型
    • key：队列和交换机绑定的RoutingKey

4. 补充 ： xml配置文件

上述通过配置类方式来配置，下面是xml配置方式下的配置文件

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
       xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
       xmlns:context="http://www.springframework.org/schema/context"
       xmlns:rabbit="http://www.springframework.org/schema/rabbit"
       xsi:schemaLocation="http://www.springframework.org/schema/beans
        http://www.springframework.org/schema/beans/spring-beans.xsd
        http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd http://www.springframework.org/schema/rabbit http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">

    <context:property-placeholder location="classpath:rabbitmq.properties"/>

    <!-- 创建连接类 连接安装好的 rabbitmq connectionFactory -->
    <bean id="connectionFactory" class="org.springframework.amqp.rabbit.connection.CachingConnectionFactory">
        <constructor-arg value="${rabbitmq.host}"/>
        <!-- username,访问RabbitMQ服务器的账户,默认是guest -->
        <property name="username" value="${rabbitmq.username}"/>
        <!-- username,访问RabbitMQ服务器的密码,默认是guest -->
        <property name="password" value="${rabbitmq.password}"/>
        <!-- host,RabbitMQ服务器地址，默认值"localhost" -->
        <property name="host" value="${rabbitmq.host}"/>
        <!-- port，RabbitMQ服务端口，默认值为5672 -->
        <property name="port" value="${rabbitmq.port}"/>
        <!-- channel-cache-size，channel的缓存数量，默认值为25 -->
        <property name="channelCacheSize" value="${rabbitmq.channelCacheSize}"/>
        <!-- cache-mode，缓存连接模式，默认值为CHANNEL(单个connection连接，连接之后关闭，自动销毁) -->
        <property name="cacheMode" value="${rabbitmq.cacheMode}"/>
    </bean>
    <!-- 创建一个Rabbit管理员，使用上述的配置-->
    <rabbit:admin connection-factory="connectionFactory"/>

    <!--或者这样配置，connection-factory元素实际就是注册一个org.springframework.amqp.rabbit.connection.CachingConnectionFactory实例
    <rabbit:connection-factory id="connectionFactory" host="${rabbitmq.host}" port="${rabbitmq.port}" username="${rabbitmq.username}" password="${rabbitmq.password}" />-->


    <!--
        定义消息队列:
        durable :   是否持久化（默认是true），如果想在RabbitMQ退出或崩溃的时候，不会失去所有的queue和消息，
                    需要同时标志队列(queue)和交换机(exchange)是持久化的，即rabbit:queue标签和rabbit:direct-exchange中的durable=true,
                    而消息(message)默认是持久化的可以看类org.springframework.amqp.core.MessageProperties中的属性
                    public static final MessageDeliveryMode DEFAULT_DELIVERY_MODE = MessageDeliveryMode.PERSISTENT;

        exclusive :  仅创建者可以使用的私有队列，断开后自动删除；

        auto_delete : 当所有消费客户端连接断开后，是否自动删除队列
     -->
    <rabbit:queue name="queueName1" id="queueName1" durable="false" auto-delete="false" exclusive="false"/>


    <!--
        绑定队列, rabbitmq的exchangeType常用的四种模式：direct，fanout，topic， headers,我们用direct模式，
        即rabbit:direct-exchange标签
            Direct交换器很简单，如果是Direct类型，就会将消息中的RoutingKey与该Exchange关联的所有Binding中的BindingKey进行比较，
            如果相等，则发送到该Binding对应的Queue中。有一个需要注意的地方：如果找不到指定的exchange，就会报错。但routing key找不到的话，不会报错，这条消息会直接丢失，所以此处要小心,
        durable : 是否持久化, 默认true
        auto-delete:自动删除，若为true，则该交换机所有队列queue删除后，自动删除交换机，默认为false
    -->
    <rabbit:direct-exchange id="exchangeKey_direct" name="exchangeKey_direct" durable="false"  auto-delete="false">
        <rabbit:bindings>
            <rabbit:binding queue="queueName1" key="routerKey1"></rabbit:binding>
        </rabbit:bindings>
    </rabbit:direct-exchange>

    <!--<rabbit:fanout-exchange id="exchangeKey_fanout" name="exchangeKey_fanout" durable="true" auto-delete="false">-->
        <!--<rabbit:bindings>-->
            <!--<rabbit:binding queue="queueKey"></rabbit:binding>-->
        <!--</rabbit:bindings>-->
    <!--</rabbit:fanout-exchange>-->

    <!-- spring amqp默认的是jackson 的一个插件,目的将生产者生产的数据转换为json存入消息队列，由于fastjson的速度快于jackson,这里替换为fastjson的一个实现 -->
    <!-- 或者配置jackson -->
    <bean id="jsonMessageConverter" class="org.springframework.amqp.support.converter.Jackson2JsonMessageConverter"/>

    <!--<rabbit:template  id="rabbitTemplate" exchange="exchangeKey_direct" connection-factory="connectionFactory"-->
                     <!--message-converter="jsonMessageConverter" />-->

    <bean id="rabbitTemplate" class="org.springframework.amqp.rabbit.core.RabbitTemplate">
        <constructor-arg name="connectionFactory" ref="connectionFactory"></constructor-arg>
    </bean>

    <!-- 自定义消息驱动pojo类 -->
    <bean id="rabbitMQHandler" class="com.kingfish.pojo.handler.RabbitMQHandler"></bean>

    <!-- 用于消息的监听的代理类 MessageListenerAdapter -->
    <bean id="testQueueListenerAdapter" class="org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter">
        <!-- 消费者类名；defaultListenerMethod 指定消费者方法名，默认是handleMessage-->
        <constructor-arg ref="rabbitMQHandler"/>
        <property name="defaultListenerMethod" value="onMessage"/>
        <property name="messageConverter" ref="jsonMessageConverter"/>
    </bean>

    <!--
        acknowledge : 设置应答模式 -> auto、manual、none 三种应答模式
                配置监听acknowledeg="manual"设置手动应答，它能够保证即使在一个worker处理消息的时候用CTRL+C来杀掉这个worker，
                或者一个consumer挂了(channel关闭了、connection关闭了或者TCP连接断了)，也不会丢失消息。
                因为RabbitMQ知道没发送ack确认消息导致这个消息没有被完全处理，将会对这条消息做re-queue处理。
                如果此时有另一个consumer连接，消息会被重新发送至另一个consumer会一直重发,直到消息处理成功,
        concurrency : 每个侦听器最初要启动的并发使用者的数量
        max-concurrency : 每个侦听器最大并发使用者数量
    -->
    <rabbit:listener-container connection-factory="connectionFactory" acknowledge="auto" concurrency="20"
                               max-concurrency="30">
        <rabbit:listener queues="queueName1" ref="rabbitMQHandler" method="onMessage"/>
        <!-- 也可用下面这种方式-->
        <!--<rabbit:listener queues="queueName1" ref="testQueueListenerAdapter"/>-->

    </rabbit:listener-container>

</beans>
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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108

注：
配置类方式使用xml时需要在 AbstractAnnotationConfigDispatcherServletInitializer 实现类上加上注解@ImportResource 引入配置文件。否则可能无法自动注入xml中声明的bean

---

以上：内容部分参考
https://www.cnblogs.com/cjm123/p/9679171.html
https://www.cnblogs.com/zhanghaoliang/p/7886110.html
https://www.jianshu.com/p/3d43561bb3ee
https://www.cnblogs.com/vipstone/p/9275256.html
https://blog.csdn.net/qq_27384769/article/details/79615015
如有侵扰，联系删除。 内容仅用于自我记录学习使用。如有错误，欢迎指正