SpringCloud源码探析（六）-消息队列RabbitMQ_spring cloud rabbitmq

1.概述

RabbitMQ是一个开源的消息代理和队列服务器，它是基于Erlang语言开发，并且是基于AMQP协议的。由于Erlang语言最初使用与交换机领域架构，因此使得RabbitMQ在Broker之间的数据交互具有良好的性能。AMQP(Advanced Message Queuing Protocol，高级消息队列协议)是一种消息队列应用层协议，专门面向消息的中间件而设计，类似于JAVA的JMS协议，基于此规范能够开发出各种各样的消息中间件。RabbitMQ能够与SpringAMQP完美整合，具有较强的扩展性和丰富的API，且具有高可靠性和低延时的优点，被业界广泛使用。本文将介绍SpringAMQP与Rabbit的使用，帮助读者更好地理解消息队列。

2.消息队列RabbitMQ使用

2.1 MQ对比

由上图可知，从可靠性和消息延迟的角度来说，RabbitMQ都是较为突出的，从吞吐量的角度来说可能一般。因此在选用消息队列中间件时，应根据使用场景选择更为合适的。

2.2 AMQP核心概念

2.3 RabbitMQ架构图

由上图可知，生产者通过channel连接到Broker，将消息投送到Exchange，Exchange再按照指定规则将消息投送到对应的queue，消费者通过监听指定的queue来获取消息。生产者不需要关注投递到那个队列，消费者也不关心消息是从哪个Exchange发送而来，两者之间是松耦合的关系。Exchange的类型有：Fanout交换机、Direct交换机、Topic交换机。Fanout交换机是一种广播模式，消息进来时会投递到所有与之绑定的队列。Direct交换机是完全根据key进行匹配队列，key一致时才会投送。Topic交换机可以按一定的规则进行匹配key，匹配成功进行投递。

2.4 SpringBoot中使用RabbitMQ

2.4.1 引入pom文件

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-amqp</artifactId>
        </dependency>
1
2
3
4

2.4.2 添加配置

spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.virtual-host=/
spring.rabbitmq.username=admin
spring.rabbitmq.password=admin
1
2
3
4
5

2.4.3 编写生产者

@RunWith(SpringRunner.class)
@SpringBootTest(classes = RabbitApplication.class)
public class SpringAmqpTest {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSimpleQueue() {
        // 队列名称
        String queueName = "simple.queue";
        // 消息
        String message = "hello, world!";
        // 发送消息
        rabbitTemplate.convertAndSend(queueName, message);
    }

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

2.4.3 编写消费者

@Component
public class RabbitMQListener {

    @RabbitListener(queues = "simple.queue")
    public void receiveMessage(String message) {
        System.out.println("接收到消息：" + message);
    }
}
1
2
3
4
5
6
7
8

2.4.4 测试结果

运行完成测试代码后，消费者收到消息。上述代码展示的是RabbitMQ最基础的发送和接收模型，生成者将消息发送到指定队列，消费者订阅指定队列获取消息。

2.5 RabbitMQ核心消息发送模型

2.5.1 WorkQueue模型

该模型一个队列可以对应多个消费者，适用于发送者发送大量消息，容易发送积压，因此需要多个消费者来消费。

2.workQueue测试代码
测试类：

    @Test
    public void testWorkQueue() throws InterruptedException {
        // 队列名称
        String queueName = "work.queue";
        // 消息
        String message = "hello, world";

        for (int i = 1; i <= 50; i++) {
            // 发送消息
            String messageInfo = message + "..." + i;
            rabbitTemplate.convertAndSend(queueName, messageInfo);
            log.info("消息发送成功：{}", messageInfo);
            Thread.sleep(50);
        }
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

消费代码：

    @RabbitListener(queues = "work.queue")
    public void receiveWorkMessage1(String message) throws InterruptedException {
        System.out.println("消息队列1...接收到消息：" + message);
        Thread.sleep(10);
    }

    @RabbitListener(queues = "work.queue")
    public void receiveWorkMessage2(String message) throws InterruptedException {
        System.out.println("消息队列2...接收到消息：" + message);
        Thread.sleep(15);
    }
1
2
3
4
5
6
7
8
9
10
11

3.workQueue运行结果

2.5.2 FanoutExchange模型

1.FanoutExchange消息发送模型

FanoutExchanger模型比简单模型和WorkQueue模型多了一个交换机（Exchange），消息先会发送到Exchange，然后Exchange将消息路由到每一个与其绑定的queue。
2.代码配置

@Configuration
public class FanoutConfig {

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

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

    @Bean
    public Binding bindingQueue1(Queue fanoutQueue1, FanoutExchange fanoutExchange) {
        return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);
    }

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

    @Bean
    public Binding bindingQueue2(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

发送部分测试代码：

  @Test
    public void testFanout() {
        // 队列名称
        String exchangeName = "fanout.exchange";
        // 消息
        String message = "hello, world!";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName,"", message);
    }
1
2
3
4
5
6
7
8
9

3.运行结果

2.5.3 DirectExchange模型

DirectExchange的消息发送模式与FanoutExchange模型基本一致，区别在于它会将收到的消息根据规则路由到指定的Queue，也就是说中间多了一层根据规则筛选发送队列，它的每一个Queue都与Exchange设置一个BindingKey，发布者发送消息时，指定消息的RoutingKey，Exchange将消息路由到BindingKey与消息RoutingKey一致的队列。
1.代码如下
监听者部分代码如下：

 @RabbitListener(bindings = @QueueBinding(value = @Queue("direct.queue1"),exchange = @Exchange(value = "direct.exchange",type = ExchangeTypes.DIRECT),key = {"red","blue"}))
    public void receiveDirectQueue1(String message) throws InterruptedException {
        System.out.println("消息队列1...接收到消息：" + message);
    }


    @RabbitListener(bindings = @QueueBinding(value = @Queue("direct.queue2"),exchange = @Exchange(value = "direct.exchange",type = ExchangeTypes.DIRECT),key = {"red","yellow"}))
    public void receiveDirectQueue2(String message) throws InterruptedException {
        System.out.println("消息队列2...接收到消息：" + message);
    }
1
2
3
4
5
6
7
8
9
10

测试代码如下：

 @Test
    public void testDirect() {
        // 队列名称
        String exchangeName = "direct.exchange";
        // 消息
        String message = "hello, world!";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName,"red", message);
    }

    @Test
    public void testDirectYellow() {
        // 队列名称
        String exchangeName = "direct.exchange";
        // 消息
        String message = "hello, world!";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName,"yellow", message);
    }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

2.测试结果


由上述结果可知，当队列订阅同一个Exchange时，向指定Exchange并携带RoutingKey，只有包含对应RoutingKey的队列才能收到消息。

2.5.4 TopicExchange模型

TopicExchange与DirectExchange类似，区别在于RoutingKey必须是多个单词的列表，并且以.分割，Queue与Exchange指定BindingKey时可以使用通配符：

#：代指0个或多个单词
*：代指一个单词。

1.测试代码

 @RabbitListener(bindings = @QueueBinding(value = @Queue("topic.queue2"),exchange = @Exchange(value = "topic.exchange",type = ExchangeTypes.TOPIC),key = {"#.news"}))
    public void receiveTopicQueue2(String message) throws InterruptedException {
        System.out.println("消息队列2...接收到消息：" + message);
    }
1
2
3
4

发送消息部分代码：

@Test
    public void testTopicQueue() {
        // 队列名称
        String exchangeName = "topic.exchange";
        // 消息
        String message = "湖人总冠军";
        // 发送消息
        rabbitTemplate.convertAndSend(exchangeName,"sports.news", message);
    }
1
2
3
4
5
6
7
8
9

2.测试结果

3.小结

1.RabbitMQ是一个遵循AMQP协议的消息中间件，消息从生产者发送到消费者，他能根据规则指定消息发送对象，缓存或进行持久化；
2.Spring AMQP是基于AMQP协议开放的接口，能够无缝对接各种基于AMQP的协议，快速利用Spring进行开发；
3.RabbitMQ延时低，而可靠性高，适用于吞吐量较大且对信息实时性要求较高的场景。

4.参考文献

1.https://www.bilibili.com/video/BV1LQ4y127n4
2.https://juejin.cn/post/6844903903637536775
3.https://www.cnblogs.com/tinmh/p/6026726.html

5.附录

https://gitee.com/Marinc/nacos.git