Spring Cloud学习（八）【RabbitMQ消息队列 服务异步通讯】_springcloud 集成rabbitmq 并发送消息

---

初识 MQ

同步通讯

同步调用的问题

一个服务 down 掉之后会影响其它服务的执行

同步调用的优点：

• 时效性较强，可以立即得到结果

同步调用的问题：

• 耦合度高
• 性能和吞吐能力下降
• 有额外的资源消耗
• 有级联失败问题

异步通讯

异步调用常见实现就是事件驱动模式

优势一：服务解耦

优势二：性能提升，吞吐量提高

优势三：服务没有强依赖，不担心级联失败问题

优势四：流量削峰

异步通信的优点：

• 耦合度低
• 吞吐量提升
• 故障隔离
• 流量削峰

异步通信的缺点：

• 依赖于Broker的可靠性、安全性、吞吐能力
• 架构复杂了，业务没有明显的流程线，不好追踪管理

MQ 常见框架

什么是MQ

MQ （MessageQueue），中文是消息队列，字面来看就是存放消息的队列。也就是事件驱动架构中的 Broker。

RabbitMQ 快速入门

RabbitMQ是基于Erlang语言开发的开源消息通信中间件，官网地址：https://www.rabbitmq.com/

RabbitMQ 单机部署

1. 下载镜像

方式一：在线拉取

docker pull rabbitmq:3-management
1

方式二：从本地加载

上传镜像包到虚拟机中后，使用命令加载镜像即可：

docker load -i mq.tar
1

2. 安装MQ

执行下面的命令来运行MQ容器：

docker run \
 -e RABBITMQ_DEFAULT_USER=admin \
 -e RABBITMQ_DEFAULT_PASS=123 \
 --name mq \
 --hostname mq1 \
 -p 15672:15672 \
 -p 5672:5672 \
 -d \
 rabbitmq:3-management
1
2
3
4
5
6
7
8
9

通过虚拟机地址+端口，访问容器

添加用户

添加虚拟主机

用户所能访问的虚拟主机地址

RabbitMQ概述

RabbitMQ中的几个概念：

• channel：操作MQ的工具
• exchange：路由消息到队列中
• queue：缓存消息
• virtual host：虚拟主机，是对queue、exchange等资源的逻辑分组

常见消息模型

MQ的官方文档中给出了5个 MQ 的 Demo 示例，对应了几种不同的用法：

• 基本消息队列（BasicQueue）
• 工作消息队列（WorkQueue）
  

发布订阅（Publish、Subscribe），又根据交换机类型不同分为三种：

• Fanout Exchange：广播
• Direct Exchange：路由
• Topic Exchange：主题

官方的HelloWorld是基于最基础的消息队列模型来实现的，只包括三个角色：

• publisher：消息发布者，将消息发送到队列queue
• queue：消息队列，负责接受并缓存消息
• consumer：订阅队列，处理队列中的消息

消息发布者

public class PublisherTest {
    @Test
    public void testSendMessage() throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.150.101");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("itcast");
        factory.setPassword("123321");
        // 1.2.建立连接
        Connection connection = factory.newConnection();

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

        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4.发送消息
        String message = "hello, rabbitmq!";
        channel.basicPublish("", queueName, null, message.getBytes());
        System.out.println("发送消息成功：【" + message + "】");

        // 5.关闭通道和连接
        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

消息消费者

public class ConsumerTest {

    public static void main(String[] args) throws IOException, TimeoutException {
        // 1.建立连接
        ConnectionFactory factory = new ConnectionFactory();
        // 1.1.设置连接参数，分别是：主机名、端口号、vhost、用户名、密码
        factory.setHost("192.168.150.101");
        factory.setPort(5672);
        factory.setVirtualHost("/");
        factory.setUsername("itcast");
        factory.setPassword("123321");
        // 1.2.建立连接
        Connection connection = factory.newConnection();

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

        // 3.创建队列
        String queueName = "simple.queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 4.订阅消息
        channel.basicConsume(queueName, true, new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                // 5.处理消息
                String message = new String(body);
                System.out.println("接收到消息：【" + message + "】");
            }
        });
        System.out.println("等待接收消息。。。。");
    }
}
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

基本消息队列的消息发送流程：

1. 建立connection
2. 创建channel
3. 利用channel声明队列
4. 利用channel向队列发送消息

基本消息队列的消息接收流程：

1. 建立connection
2. 创建channel
3. 利用channel声明队列
4. 定义consumer的消费行为handleDelivery()
5. 利用channel将消费者与队列绑定

SpringAMQP

SpringAmqp的官方地址：https://spring.io/projects/spring-amqp

Spring AMQP 项目将Spring的核心概念应用于基于AMQP的消息传递解决方案的开发。它提供了一个“模板”，作为发送和接收消息的高级抽象。它还通过“侦听器容器”为消息驱动的POJO提供支持。这些库促进了AMQP资源的管理，同时促进了依赖注入和声明性配置的使用。在所有这些情况下，您将看到与Spring框架中JMS支持的相似之处。

该项目由两部分组成；spring-amqp 是基础抽象，spring-rabbit 是RabbitMQ 实现。

功能

• 用于异步处理入站消息的侦听器容器
• 用于发送和接收消息的RabbitTemplate
• 用于自动声明队列、交换和绑定的RabbitAdmin

SimpleQueue 模型

利用SpringAMQP实现HelloWorld中的基础消息队列功能

1. 引入AMQP依赖

因为publisher和consumer服务都需要amqp依赖，因此这里把依赖直接放到父工程mq-demo中：

<!--AMQP依赖，包含RabbitMQ-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
1
2
3
4
5

2. 在publisher中编写测试方法，向simple.queue发送消息

在publisher服务中编写application.yml，添加mq连接信息：

spring:
  rabbitmq:
    host: 10.211.55.6 # rabbitMQ的ip
    port: 5672 # 端口
    username: itcast
    password: 123321
    virtual-host: / # 虚拟主机
1
2
3
4
5
6
7

在publisher服务中新建一个测试类，编写测试方法：

@RunWith(SpringRunner.class)
@SpringBootTest
public class SpringAmqpTest {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSendMessage2SimpleQueue(){
        String queueName = "simple.queue";
        String message = "hello, spring ampq!";
        // 这个方法使用前要确保队列已经被创建
        rabbitTemplate.convertAndSend(queueName, message);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

运行程序后，可以看到队列中多了一条消息

什么是AMQP？

• 应用间消息通信的一种协议，与语言和平台无关。

SpringAMQP如何发送消息？

• 引入 amqp 的 starter 依赖
• 配置 RabbitMQ 地址
• 利用 RabbitTemplate的convertAndSend 方法

3. 在 consumer 中编写消费逻辑，监听 simple.queue

在consumer服务中编写application.yml，添加mq连接信息：

spring:
  rabbitmq:
    host: 10.211.55.6 # rabbitMQ的ip
    port: 5672 # 端口
    username: itcast
    password: 123321
    virtual-host: / # 虚拟主机
1
2
3
4
5
6
7

在consumer服务中新建一个类，编写消费逻辑：

@Component
public class SpringRabbitListener {

    @RabbitListener(queues = "simple.queue")  // 指定监听队列
    public void listenSimpleQueue(String msg){
        System.out.println("消费者接收到simple.queue的消息：【" + msg + "】");
    }
}
1
2
3
4
5
6
7
8

运行启动类

SpringAMQP如何接收消息？

• 引入 amqp 的 starter 依赖
• 配置 RabbitMQ 地址
• 定义类，添加 @Component 注解
• 类中声明方法，添加 @RabbitListener 注解，方法参数就时消息

注意： 消息一旦消费就会从队列删除，RabbitMQ 没有消息回溯功能

WorkQueue 模型

案例：模拟WorkQueue，实现一个队列绑定多个消费者

基本思路如下：

1. 在publisher服务中定义测试方法，每秒产生50条消息，发送到simple.queue

@Test
public void testSendMessage2WorkQueue() throws InterruptedException {
    String queueName = "simple.queue";
    String message = "hello, message__";
    for (int i = 0; i < 50; i++) {
        rabbitTemplate.convertAndSend(queueName, message + i);
        Thread.sleep(20);
    }
}
1
2
3
4
5
6
7
8
9

2. 在consumer服务中定义两个消息监听者，都监听simple.queue队列

@RabbitListener(queues = "simple.queue")
public void listenWorkQueue1(String msg) throws InterruptedException {
    System.out.println("消费者1接收到simple.queue的消息：【" + msg + "】" + LocalTime.now());
    Thread.sleep(20);
}

@RabbitListener(queues = "simple.queue")
public void listenWorkQueue2(String msg) throws InterruptedException {
    System.err.println("消费者2....接收到simple.queue的消息：【" + msg + "】" + LocalTime.now());
    Thread.sleep(200);
}
1
2
3
4
5
6
7
8
9
10
11

消费预取限制

3. 消费者1每秒处理50条消息，消费者2每秒处理10条消息

spring:
  rabbitmq:
    host: 10.211.55.6 # rabbitMQ的ip
    port: 5672 # 端口
    username: itcast
    password: 123321
    virtual-host: / # 虚拟主机
    listener:
      simple:
        prefetch: 1  # 每次只能获取一条消息，处理完成才能获取下一个消息
1
2
3
4
5
6
7
8
9
10

Work模型的使用：

• 多个消费者绑定到一个队列，同一条消息只会被一个消费者处理
• 通过设置prefetch来控制消费者预取的消息数量

发布订阅模型

发布订阅模式与之前案例的区别就是允许将同一消息发送给多个消费者。实现方式是加入了exchange（交换机）。
常见exchange类型包括：

• Fanout：广播
• Direct：路由
• Topic：话题

注意： exchange负责消息路由，而不是存储，路由失败则消息丢失

发布订阅-Fanout Exchange

Fanout Exchange 会将接收到的消息广播到每一个跟其绑定的queue

利用SpringAMQP演示FanoutExchange的使用

实现思路如下：

1. 在consumer服务中，利用代码声明队列、交换机，并将两者绑定
2. 在consumer服务中，编写两个消费者方法，分别监听fanout.queue1和fanout.queue2
3. 在publisher中编写测试方法，向itcast.fanout发送消息

1. 步骤1：在consumer服务声明Exchange、Queue、Binding

SpringAMQP提供了声明交换机、队列、绑定关系的API，例如：

在consumer服务常见一个类，添加@Configuration注解，并声明FanoutExchange、Queue和绑定关系对象Binding，代码如下：

@Configuration
public class FanoutConfig {

    // itcast.fanout
    @Bean
    public FanoutExchange fanoutExchange(){
        return new FanoutExchange("itcast.fanout");
    }

    // fanout.queue1
    @Bean
    public Queue fanoutQueue1(){
        return new Queue("fanout.queue1");
    }

    // 将队列1绑定到交换机
    @Bean
    public Binding fanoutBinding1(Queue fanoutQueue1, FanoutExchange fanoutExchange){
        return BindingBuilder
                .bind(fanoutQueue1)
                .to(fanoutExchange);
    }

    // fanout.queue2
    @Bean
    public Queue fanoutQueue2(){
        return new Queue("fanout.queue2");
    }

    // 将队列2绑定到交换机
    @Bean
    public Binding fanoutBinding2(Queue fanoutQueue2, FanoutExchange fanoutExchange){
        return BindingBuilder
                .bind(fanoutQueue2)
                .to(fanoutExchange);
    }
}
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

2. 步骤2：在consumer服务声明两个消费者

在consumer服务的SpringRabbitListener类中，添加两个方法，分别监听fanout.queue1和fanout.queue2：

@Component
public class SpringRabbitListener {

    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue1(String msg) throws InterruptedException {
        System.out.println("消费者1接收到simple.queue的消息：【" + msg + "】" + LocalTime.now());
        Thread.sleep(20);
    }

    @RabbitListener(queues = "simple.queue")
    public void listenWorkQueue2(String msg) throws InterruptedException {
        System.err.println("消费者2....接收到simple.queue的消息：【" + msg + "】" + LocalTime.now());
        Thread.sleep(200);
    }

    @RabbitListener(queues = "fanout.queue1")
    public void listenFanoutQueue1(String msg) throws InterruptedException {
        System.out.println("消费者接收到fanout.queue1的消息：【" + msg + "】" + LocalTime.now());
    }

    @RabbitListener(queues = "fanout.queue2")
    public void listenFanoutQueue2(String msg) throws InterruptedException {
        System.out.println("消费者接收到fanout.queue2的消息：【" + msg + "】" + LocalTime.now());
    }

}
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

3. 步骤3：在publisher服务发送消息到FanoutExchange

在publisher服务的SpringAmqpTest类中添加测试方法：

@Test
public void testSendFanoutExchange(){

    // 交换机名称
    String exchangeName = "itcast.fanout";
    // 消息
    String message = "hello, every one!";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "", message);
}
1
2
3
4
5
6
7
8
9
10

运行结果：

交换机的作用是什么？

• 接收publisher发送的消息
• 将消息按照规则路由到与之绑定的队列
• 不能缓存消息，路由失败，消息丢失
• FanoutExchange的会将消息路由到每个绑定的队列

声明队列、交换机、绑定关系的Bean是什么？

• Queue
• FanoutExchange
• Binding

发布订阅-DirectExchange

Direct Exchange 会将接收到的消息根据规则路由到指定的Queue，因此称为路由模式（routes）。

• 每一个Queue都与Exchange设置一个BindingKey
• 发布者发送消息时，指定消息的RoutingKey
• Exchange将消息路由到BindingKey与消息RoutingKey一致的队列

利用SpringAMQP演示DirectExchange的使用

实现思路如下：

1. 利用@RabbitListener声明Exchange、Queue、RoutingKey
2. 在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2
3. 在publisher中编写测试方法，向itcast. direct发送消息

步骤1：在consumer服务声明Exchange、Queue

1. 在consumer服务中，编写两个消费者方法，分别监听direct.queue1和direct.queue2，
2. 并利用@RabbitListener声明Exchange、Queue、RoutingKey

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "direct.queue1"),
        exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
        key = {"red", "blue"}
))
public void listenDirectQueue1(String msg) throws InterruptedException {
    System.out.println("消费者接收到direct.queue1的消息：【" + msg + "】" + LocalTime.now());
}

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "direct.queue2"),
        exchange = @Exchange(name = "itcast.direct", type = ExchangeTypes.DIRECT),
        key = {"red", "yellow"}
))
public void listenDirectQueue2(String msg) throws InterruptedException {
    System.out.println("消费者接收到direct.queue2的消息：【" + msg + "】" + LocalTime.now());
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

步骤2：在publisher服务发送消息到DirectExchange

在publisher服务的SpringAmqpTest类中添加测试方法：

@Test
public void testSendDirectExchange(){

    // 交换机名称
    String exchangeName = "itcast.direct";
    // 消息
    String message = "hello, red!";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "red", message);
}
1
2
3
4
5
6
7
8
9
10

运行结果

描述下Direct交换机与Fanout交换机的差异？

• Fanout交换机将消息路由给每一个与之绑定的队列
• Direct交换机根据RoutingKey判断路由给哪个队列
• 如果多个队列具有相同的RoutingKey，则与Fanout功能类似

基于@RabbitListener注解声明队列和交换机有哪些常见注解？

• @Queue
• @Exchange
• @QueueBinding

发布订阅-TopicExchange

TopicExchange与DirectExchange类似，区别在于routingKey必须是多个单词的列表，并且以 . 分割。

Queue与Exchange指定BindingKey时可以使用通配符：
#：代指0个或多个单词
*：代指一个单词

利用SpringAMQP演示TopicExchange的使用

实现思路如下：

1. 并利用@RabbitListener声明Exchange、Queue、RoutingKey
2. 在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2
3. 在publisher中编写测试方法，向itcast. topic发送消息

步骤1：在consumer服务声明Exchange、Queue

1. 在consumer服务中，编写两个消费者方法，分别监听topic.queue1和topic.queue2，
2. 并利用@RabbitListener声明Exchange、Queue、RoutingKey

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "topic.queue1"),
        exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
        key = "china.#"
))
public void listenTopicQueue1(String msg) throws InterruptedException {
    System.out.println("消费者接收到topic.queue1的消息：【" + msg + "】" + LocalTime.now());
}

@RabbitListener(bindings = @QueueBinding(
        value = @Queue(name = "topic.queue2"),
        exchange = @Exchange(name = "itcast.topic", type = ExchangeTypes.TOPIC),
        key = "#.news"
))
public void listenTopicQueue2(String msg) throws InterruptedException {
    System.out.println("消费者接收到topic.queue2的消息：【" + msg + "】" + LocalTime.now());
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

步骤2：在publisher服务发送消息到TopicExchange

在publisher服务的SpringAmqpTest类中添加测试方法：

@Test
public void testSendTopicExchange(){

    // 交换机名称
    String exchangeName = "itcast.topic";
    // 消息
    String message = "天气新闻!!!";
    // 发送消息
    rabbitTemplate.convertAndSend(exchangeName, "china.weather", message);
}
1
2
3
4
5
6
7
8
9
10

运行结果：

描述下Direct交换机与Topic交换机的差异？

• Topic交换机接收的消息RoutingKey必须是多个单词，以 . 分割
• Topic交换机与队列绑定时的bindingKey可以指定通配符
• #：代表0个或多个词
• *：代表1个词

消息转换器

测试发送Object类型消息

说明：在SpringAMQP的发送方法中，接收消息的类型是Object，也就是说我们可以发送任意对象类型的消息，SpringAMQP会帮我们序列化为字节后发送。

我们在consumer中利用@Bean声明一个队列：

@Bean
public Queue objectQueue(){
    return new Queue("object.queue");
}
1
2
3
4

在publisher中发送消息以测试：

@Test
public void testSendObjectQueue(){

    Map<String, Object> msg = new HashMap<>();
    msg.put("name", "wxx");
    msg.put("age", "18");
    rabbitTemplate.convertAndSend("object.queue", msg);
}
1
2
3
4
5
6
7
8

Spring的对消息对象的处理是由org.springframework.amqp.support.converter.MessageConverter来处理的。而默认实现是SimpleMessageConverter，基于JDK的ObjectOutputStream完成序列化。
如果要修改只需要定义一个MessageConverter 类型的Bean即可。推荐用JSON方式序列化，步骤如下：

• 我们在publisher服务引入依赖

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>
1
2
3
4

• 我们在publisher服务声明MessageConverter：

@Bean
public MessageConverter messageConverter(){
    return new Jackson2JsonMessageConverter();
}
1
2
3
4

我们在consumer服务引入Jackson依赖：

<dependency>
    <groupId>com.fasterxml.jackson.core</groupId>
    <artifactId>jackson-databind</artifactId>
</dependency>
1
2
3
4

我们在consumer服务定义MessageConverter：

@Bean
public MessageConverter messageConverter(){
    return new Jackson2JsonMessageConverter();
}
1
2
3
4

然后定义一个消费者，监听object.queue队列并消费消息：

@RabbitListener(queues = "object.queue")
public void listenObjectQueue(Map<String, Object> msg){
    System.out.println("接收到object.queue的消息：" + msg);
}
1
2
3
4

运行结果

SpringAMQP中消息的序列化和反序列化是怎么实现的？

• 利用MessageConverter实现的，默认是JDK的序列化
• 注意发送方与接收方必须使用相同的MessageConverter