RabbitMQ笔记

旧城等待， 2022-02-09 13:29 188阅读 0赞

# RabbitMQ笔记 #

参考资料：

*  [Spring Boot(八)：RabbitMQ 详解][Spring Boot_RabbitMQ]
 *  [我为什么要选择RabbitMQ ，RabbitMQ简介，各种MQ选型对比][RabbitMQ _RabbitMQ_MQ]
 *  [RabbitMQ技术详解][RabbitMQ]
 *  [RabbitMQ 中文文档][RabbitMQ 1]
 *  [rabbitmq消息队列的简单入门][rabbitmq]
 *  [RabbitMQ教程][RabbitMQ 2]
 *  [RabbitMQ核心概念篇][RabbitMQ 3]

## 前言 ##

在大多应用中，我们系统之间需要进行异步通信，即异步消息。异步消息中两个重要概念:消息代理(message broker)和目的地(destination)。当消息发送者发送消息以后，将由消息代理接管，消息代理保证消息传递到指定目的地。

异步消息主要有两种形式的目的地:

1.  队列(queue):点对点消息通信(point-to-point)
    
     *  消息发送者发送消息，消息代理将其放入一个队列中，消息接收者从队列中获取消息内容，消息读取后被移出队列
     *  消息只有唯一的发送者和接收者，但并不是说只能有一个接收者
2.  主题(topic):发布(publish)/订阅(subscribe)消息通信
    
     *  发送者(发布者)发送消息到主题，多个接收者(订阅者)监听(订阅)这个主题，那么 就会在消息到达时同时收到消息

消息代理实现：

1.  JMS(Java Message Service)java消息服务:
    
     *  基于JVM消息代理的规范。ActiveMQ、 HornetMQ是JMS实现
2.  AMQP(Advanced Message Queuing Protocol)
    
     *  高级消息队列协议，也是一个消息代理的规范，兼容JMS
     *  RabbitMQ是AMQP的实现

AMQP，即 Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。在分布式的系统中，消息队列也会被用在很多其它的方面，比如：分布式事务的支持，RPC 的调用等等。AMQP 的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。

关于消息队列的详细介绍请参阅：

*  [《Java帝国之消息队列》][Java]
 *  [《一个故事告诉你什么是消息队列》][Link 1]
 *  [《到底什么时候该使用MQ》][MQ]

## 简介 ##

RabbitMQ 是一个开源的 AMQP 实现，服务器端用Erlang语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP 等，支持 AJAX。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

![831021-20170903123136077-267287773.png][]

## **核心概念** ##

**RabbitMQ运行机制**  
![RabbitMQ运行机制][RabbitMQ 4]

1.  Producer&Consumer
    
     *  producer指的是消息生产者，consumer消息的消费者。
     *  它提供一种传输服务,它的角色就是维护一条从生产者到消费者的路线，保 证数据能按照指定的方式进行传输
2.  Queue
    
     *  消息队列，提供了FIFO的处理机制，具有缓存消息的能力。rabbitmq中， 队列消息可以设置为持久化，临时或者自动删除。
        
         *  设置为持久化的队列，queue中的消息会在server本地硬盘存储一份，防止 系统crash，数据丢失
         *  设置为临时队列，queue中的数据在系统重启之后就会丢失
         *  设置为自动删除的队列，当不存在用户连接到server，队列中的数据会被自  
            动删除
     *  RabbitMQ中的消息都只能存储在Queue中，生产者生产消息并最终投递到Queue中，可以从Queue中获取消息并消费。
     *  多个消费者可以订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。
3.  Exchange
    
     *  消息交换机,它指定消息按什么规则,路由到哪个队列。
     *  Exchange有4种类型:direct(默认)，fanout, topic, 和headers，不同类型 的Exchange转发消息的策略有所区别。具体见下文。
4.  Binding
    
     *  将一个特定的 Exchange 和一个特定的 Queue 绑定起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。
     *  Exchange 和Queue的绑定可以是多对多的关系。
     *  在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding  
        key；消费者将消息发送给Exchange时，一般会指定一个routing key；当binding key与routing key相匹配时，消息将会被路由到对应的Queue中。
     *  binding key 并不是在所有情况下都生效，它依赖于Exchange Type，比如fanout类型的Exchange就会无视binding key，而是将消息路由到所有绑定到该Exchange的Queue。
5.  virtual host( vhosts )
    
     *  在rabbitmq server上可以创建多个虚拟的message broker，又叫做 virtual hosts (vhosts)
     *  每一个vhost本质上是一个mini-rabbitmq server，分别管理各自的 exchange和bindings
     *  vhost相当于物理的server，可以为不同app提供边界隔离
     *  producer和consumer连接rabbit server需要指定一个vhost

### Exchange类型 ###

#### 1. fanout ####

fanout类型的Exchange路由规则会把所有发送到该Exchange的消息路由到所有与它绑定的Queue中。

![在这里插入图片描述][20190503115044996.png]

上图中，生产者（P）发送到Exchange（X）的所有消息都会路由到图中的两个Queue，并最终被两个消费者（C1与C2）消费。

#### 2. direct ####

direct类型的Exchange路由规则会把消息路由到那些binding key与routing key完全匹配的Queue中。

![在这里插入图片描述][20190503115107276.png]

以上图的配置为例，我们以routingKey=”error”发送消息到Exchange，则消息会路由到Queue1（amqp.gen-S9b…，这是由RabbitMQ自动生成的Queue名称）和Queue2（amqp.gen-Agl…）；如果我们以routingKey=”info”或routingKey=”warning”来发送消息，则消息只会路由到Queue2。如果我们以其他routingKey发送消息，则消息不会路由到这两个Queue中。

#### 3. topic ####

前面讲到direct类型的Exchange路由规则是完全匹配binding key与routing  
key，但这种严格的匹配方式在很多情况下不能满足实际业务需求。topic类型的Exchange在匹配规则上进行了扩展，它与direct类型的Exchage相似，也是将消息路由到binding key与routing key相匹配的Queue中，但这里的匹配规则有些不同，它约定：

*  routing key为一个句点号“. ”分隔的字符串（我们将被句点号“. ”分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit”
 *  binding key与routing key一样也是句点号“. ”分隔的字符串
 *  binding key中可以存在两种特殊字符“`*`”与“\#”，用于做模糊匹配，其中“`*`”用于匹配一个单词，“\#”用于匹配多个单词（可以是零个）

![在这里插入图片描述][20190503115142203.png]

以上图中的配置为例，routingKey=”quick.orange.rabbit”的消息会同时路由到Q1与Q2，routingKey=”lazy.orange.fox”的消息会路由到Q1与Q2，routingKey=”lazy.brown.fox”的消息会路由到Q2，routingKey=”lazy.pink.rabbit”的消息会路由到Q2（只会投递给Q2一次，虽然这个routingKey与Q2的两个bindingKey都匹配）；routingKey=”quick.brown.fox”、routingKey=”orange”、routingKey=”quick.orange.male.rabbit”的消息将会被丢弃，因为它们没有匹配任何bindingKey。

#### 4. headers ####

headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。

*  在绑定Queue与Exchange时指定一组键值对；
 *  当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对；
 *  如果完全匹配则消息会路由到该Queue，否则不会路由到该Queue。

### RPC ###

MQ本身是基于异步的消息处理，前面的示例中所有的生产者（P）将消息发送到RabbitMQ后不会知道消费者（C）处理成功或者失败（甚至连有没有消费者来处理这条消息都不知道）。

但实际的应用场景中，我们很可能需要一些同步处理，需要同步等待服务端将我的消息处理完成后再进行下一步处理。这相当于RPC（Remote Procedure Call，远程过程调用）。在RabbitMQ中也支持RPC。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ZpZ2h0aW5nX05vMQ_size_16_color_FFFFFF_t_70]

[RabbitMQ][RabbitMQ 5] 中实现`RPC` 的机制是：

1.  客户端发送请求（消息）时，在消息的属性（`MessageProperties` ，在`AMQP` 协议中定义了14种`properties` ，这些属性会随着消息一起发送）中设置两个值`replyTo` （一个`Queue` 名称，用于告诉服务器处理完成后将通知我的消息发送到这个`Queue` 中）和`correlationId` （此次请求的标识号，服务器处理完成后需要将此属性返还，客户端将根据这个id了解哪条请求被成功执行了或执行失败）
2.  服务器端收到消息并处理
3.  服务器端处理完消息后，将生成一条应答消息到`replyTo` 指定的`Queue` ，同时带上`correlationId` 属性
4.  客户端之前已订阅`replyTo` 指定的`Queue` ，从中收到服务器的应答消息后，根据其中的`correlationId` 属性分析哪条请求被执行了，根据执行结果进行后续业务处理。

## [RabbitMQ实战（Python）][RabbitMQ_Python] ##

## [RabbitMQ实战（Java）][RabbitMQ_Java] ##

## Spring Boot 集成 RabbitMQ ##

Spring Boot 集成 RabbitMQ 非常简单，如果只是简单的使用配置非常少，Spring Boot 提供了`spring-boot-starter-amqp` 项目对消息各种支持。

### 简单使用 ###

1、配置 Pom 包，主要是添加 `spring-boot-starter-amqp` 的支持

<dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>

2、配置文件

配置 RabbitMQ 的安装地址、端口以及账户信息

spring.application.name=Spring-boot-rabbitmq
    
    spring.rabbitmq.host=192.168.0.86
    spring.rabbitmq.port=5672
    spring.rabbitmq.username=admin
    spring.rabbitmq.password=123456

3、队列配置

@Configuration
    public class RabbitConfig {
    
        @Bean
        public Queue Queue() {
            return new Queue("hello");
        }
    
    }

3、发送者

rabbitTemplate 是 Spring Boot 提供的默认实现

@component
    public class HelloSender {
    
        @Autowired
        private AmqpTemplate rabbitTemplate;
    
        public void send() {
            String context = "hello " + new Date();
            System.out.println("Sender : " + context);
            this.rabbitTemplate.convertAndSend("hello", context);
        }
    
    }

4、接收者

@Component
    @RabbitListener(queues = "hello")
    public class HelloReceiver {
    
        @RabbitHandler
        public void process(String hello) {
            System.out.println("Receiver  : " + hello);
        }
    
    }

5、测试

@RunWith(SpringRunner.class)
    @SpringBootTest
    public class RabbitMqHelloTest {
    
        @Autowired
        private HelloSender helloSender;
    
        @Test
        public void hello() throws Exception {
            helloSender.send();
        }
    
    }

> 注意，发送者和接收者的 queue name 必须一致，不然不能接收

### 多对多使用（work模式） ###

工作队列（又称：任务队列——Task Queues）是为了避免等待一些占用大量资源、时间的操作。当我们把任务（Task）当作消息发送到队列中，一个运行在后台的工作者（worker）进程就会取出任务然后处理。当你运行多个工作者（workers），任务就会在它们之间共享。

#### **一对多发送** ####

对上面的代码进行了小改造，接收端注册了两个 Receiver,Receiver1 和 Receiver2，发送端加入参数计数，接收端打印接收到的参数，下面是测试代码，发送一百条消息，来观察两个接收端的执行效果

@Test
    public void oneToMany() throws Exception {
        for (int i=0;i<100;i++){
            neoSender.send(i);
        }
    }

结果如下：

Receiver 1: Spring boot neo queue ****** 11
    Receiver 2: Spring boot neo queue ****** 12
    Receiver 2: Spring boot neo queue ****** 14
    Receiver 1: Spring boot neo queue ****** 13
    Receiver 2: Spring boot neo queue ****** 15
    Receiver 1: Spring boot neo queue ****** 16
    Receiver 1: Spring boot neo queue ****** 18
    Receiver 2: Spring boot neo queue ****** 17
    Receiver 2: Spring boot neo queue ****** 19
    Receiver 1: Spring boot neo queue ****** 20

根据返回结果得到以下结论

> 一个发送者，N个接受者,经过测试会均匀的将消息发送到N个接收者中

#### **多对多发送** ####

复制了一份发送者，加入标记，在一百个循环中相互交替发送

@Test
        public void manyToMany() throws Exception {
            for (int i=0;i<100;i++){
                neoSender.send(i);
                neoSender2.send(i);
            }
    }

结果如下：

Receiver 1: Spring boot neo queue ****** 20
    Receiver 2: Spring boot neo queue ****** 20
    Receiver 1: Spring boot neo queue ****** 21
    Receiver 2: Spring boot neo queue ****** 21
    Receiver 1: Spring boot neo queue ****** 22
    Receiver 2: Spring boot neo queue ****** 22
    Receiver 1: Spring boot neo queue ****** 23
    Receiver 2: Spring boot neo queue ****** 23
    Receiver 1: Spring boot neo queue ****** 24
    Receiver 2: Spring boot neo queue ****** 24
    Receiver 1: Spring boot neo queue ****** 25
    Receiver 2: Spring boot neo queue ****** 25

> 结论：和一对多一样，接收端仍然会均匀接收到消息

### 高级使用 ###

#### **对象的支持** ####

Spring Boot 以及完美的支持对象的发送和接收，不需要格外的配置。

//发送者
    public void send(User user) {
        System.out.println("Sender object: " + user.toString());
        this.rabbitTemplate.convertAndSend("object", user);
    }
    
    ...
    
    //接收者
    @RabbitHandler
    public void process(User user) {
        System.out.println("Receiver object : " + user);
    }

结果如下：

Sender object: User{name='neo', pass='123456'}
    Receiver object : User{name='neo', pass='123456'}

#### **Topic Exchange** ####

topic 是 RabbitMQ 中最灵活的一种方式，可以根据 routing\_key 自由的绑定不同的队列

首先对 topic 规则配置，这里使用两个队列来测试

@Configuration
    public class TopicRabbitConfig {
    
        final static String message = "topic.message";
        final static String messages = "topic.messages";
    
        @Bean
        public Queue queueMessage() {
            return new Queue(TopicRabbitConfig.message);
        }
    
        @Bean
        public Queue queueMessages() {
            return new Queue(TopicRabbitConfig.messages);
        }
    
        @Bean
        TopicExchange exchange() {
            return new TopicExchange("exchange");
        }
    
        @Bean
        Binding bindingExchangeMessage(Queue queueMessage, TopicExchange exchange) {
            return BindingBuilder.bind(queueMessage).to(exchange).with("topic.message");
        }
    
        @Bean
        Binding bindingExchangeMessages(Queue queueMessages, TopicExchange exchange) {
            return BindingBuilder.bind(queueMessages).to(exchange).with("topic.#");
        }
    }

使用 queueMessages 同时匹配两个队列，queueMessage 只匹配 “topic.message” 队列

public void send1() {
        String context = "hi, i am message 1";
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("exchange", "topic.message", context);
    }
    
    public void send2() {
        String context = "hi, i am messages 2";
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("exchange", "topic.messages", context);
    }

发送send1会匹配到topic.\#和topic.message 两个Receiver都可以收到消息，发送send2只有topic.\#可以匹配所有只有Receiver2监听到消息

#### **Fanout Exchange** ####

Fanout 就是我们熟悉的广播模式或者订阅模式，给 Fanout 交换机发送消息，绑定了这个交换机的所有队列都收到这个消息。

Fanout 相关配置

@Configuration
    public class FanoutRabbitConfig {
    
        @Bean
        public Queue AMessage() {
            return new Queue("fanout.A");
        }
    
        @Bean
        public Queue BMessage() {
            return new Queue("fanout.B");
        }
    
        @Bean
        public Queue CMessage() {
            return new Queue("fanout.C");
        }
    
        @Bean
        FanoutExchange fanoutExchange() {
            return new FanoutExchange("fanoutExchange");
        }
    
        @Bean
        Binding bindingExchangeA(Queue AMessage,FanoutExchange fanoutExchange) {
            return BindingBuilder.bind(AMessage).to(fanoutExchange);
        }
    
        @Bean
        Binding bindingExchangeB(Queue BMessage, FanoutExchange fanoutExchange) {
            return BindingBuilder.bind(BMessage).to(fanoutExchange);
        }
    
        @Bean
        Binding bindingExchangeC(Queue CMessage, FanoutExchange fanoutExchange) {
            return BindingBuilder.bind(CMessage).to(fanoutExchange);
        }
    
    }

这里使用了 A、B、C 三个队列绑定到 Fanout 交换机上面，发送端的 routing\_key 写任何字符都会被忽略：

public void send() {
        String context = "hi, fanout msg ";
        System.out.println("Sender : " + context);
        this.rabbitTemplate.convertAndSend("fanoutExchange","", context);
    }

结果如下：

Sender : hi, fanout msg 
    ...
    fanout Receiver B: hi, fanout msg 
    fanout Receiver A  : hi, fanout msg 
    fanout Receiver C: hi, fanout msg

结果说明，绑定到 fanout 交换机上面的队列都收到了消息

[Spring Boot_RabbitMQ]: https://www.cnblogs.com/ityouknow/p/6120544.html
[RabbitMQ _RabbitMQ_MQ]: https://www.sojson.com/blog/48.html
[RabbitMQ]: http://www.uml.org.cn/zjjs/201805234.asp
[RabbitMQ 1]: http://rabbitmq.mr-ping.com/
[rabbitmq]: https://blog.csdn.net/jsaber/article/details/79027314
[RabbitMQ 2]: https://blog.csdn.net/hellozpc/article/details/81436980
[RabbitMQ 3]: https://www.cnblogs.com/luotianshuai/p/7469365.html
[Java]: https://mp.weixin.qq.com/s?__biz=MzAxOTc0NzExNg==&mid=2665513507&idx=1&sn=d6db79c1ae03ba9260fb0fb77727bb54&chksm=80d67a60b7a1f376e7ad1e2c3276e8b565f045b1c7e21ef90926f69d99f969557737eb5d8128&mpshare=1&scene=1&srcid=1019awkBx8kaLyFohcuW4Ee7
[Link 1]: https://github.com/jasonGeng88/blog/blob/master/201705/MQ.md
[MQ]: https://mp.weixin.qq.com/s?__biz=MjM5ODYxMDA5OQ==&mid=2651960012&idx=1&sn=c6af5c79ecead98daa4d742e5ad20ce5&chksm=bd2d07108a5a8e0624ae6ad95001c4efe09d7ba695f2ddb672064805d771f3f84bee8123b8a6&mpshare=1&scene=1&srcid=04054h4e90lz5Qc2YKnLNuvY
[831021-20170903123136077-267287773.png]: /images/20220206/7c8ab8fc0015470991a294e9819bed8f.png
[RabbitMQ 4]: /images/20220206/fef4d83a4c734ba5a9ad9cdaf57ea713.png
[20190503115044996.png]: /images/20220206/92e8fe015991472791f7f62c7a9ff3a2.png
[20190503115107276.png]: /images/20220206/374b7a4ced0f4d1da75faf8797b27ae8.png
[20190503115142203.png]: /images/20220206/c504e82d82cd4f069695b412d7c7f4bb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ZpZ2h0aW5nX05vMQ_size_16_color_FFFFFF_t_70]: /images/20220206/668bf67de0934a98be13acda313cc90c.png
[RabbitMQ 5]: http://www.sojson.com/tag_rabbitmq.html
[RabbitMQ_Python]: http://rabbitmq.mr-ping.com/tutorials_with_python/%5B1%5DHello_World.html
[RabbitMQ_Java]: https://blog.csdn.net/hellozpc/article/details/81436980#5_192