RabbitMQ学习理解

蔚落 2022-05-27 11:46 129阅读 0赞

消息（Message）是指在应用间传送的数据。

消息队列（Message Queue）是一种应用间的通信方式。队列我们也可以理解为管道，以管道的方式做消息传递，消息发送后可以立即返回，由消息队列来确保消息的可靠传递。消息发布者只管将消息发布到 MQ ，消息消费者只管从 MQ 中取消息。

场景：

1.和我们平时的物流运输一样，当全国各地的快递到达快递站点后，并不可能立即到达客户手中，而是将快递依次分区域分发并派送。这也是队列机制。

即，排队。一个接着一个的处理，不能插队。

## RabbitMQ简介 ##

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。MQ主要用于异步处理、应用解耦、流量削峰。 AMQP的主要特征是面向消息、队列、路由（包括点对点和发布/订阅）、可靠性、安全。 RabbitMQ是一个开源的AMQP实现，服务器端用Erlang语言编写，支持多种客户端，如：Java。用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。

### ConnectionFactory、Connection、Channel ###

ConnectionFactory、Connection、Channel都是RabbitMQ对外提供的API中最基本的对象。Connection是RabbitMQ的socket链接，它封装了socket协议相关部分逻辑。

ConnectionFactory为Connection的制造工厂。

Channel是我们与RabbitMQ打交道的最重要的一个接口，我们大部分的业务操作是在Channel这个接口中完成的，包括定义Queue、定义Exchange、绑定Queue与Exchange、发布消息等。

### Queue ###

Queue（队列）是RabbitMQ的内部对象，用于存储消息，用下图表示。

![70][]

RabbitMQ中的消息都只能存储在Queue中，生产者（下图中的P）生产消息并最终投递到Queue中，消费者（下图中的C）可以从Queue中获取消息并消费。

![20180417184758767][]

多个消费者可以订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。

![20180417191719289][]

/**
         * 声明接收队列
         *
         * @return
         */
        @Bean
        public Queue rabbitQueue() {
            return new Queue(RECEIVE_QUEUE, true, false, false, null);
        }

### Message acknowledgment ###

在实际应用中，可能会发生消费者收到Queue中的消息，但没有处理完成就宕机（或出现其他意外）的情况，这种情况下就可能会导致消息丢失。为了避免这种情况发生，我们可以要求消费者在消费完消息后发送一个回执给RabbitMQ，RabbitMQ收到消息回执（Message acknowledgment）后才将该消息从Queue中移除；如果RabbitMQ没有收到回执并检测到消费者的RabbitMQ连接断开，则RabbitMQ会将该消息发送给其他消费者（如果存在多个消费者）进行处理。这里不存在timeout概念，一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者，除非它的RabbitMQ连接断开。 这里会产生另外一个问题，如果我们的开发人员在处理完业务逻辑后，忘记发送回执给RabbitMQ，这将会导致严重的bug——Queue中堆积的消息会越来越多；消费者重启后会重复消费这些消息并重复执行业务逻辑…

另外pub message是没有ack的。

### Message durability ###

如果我们希望即使在RabbitMQ服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置为可持久化的（durable），这样可以保证绝大部分情况下我们的RabbitMQ消息不会丢失。但依然解决不了小概率丢失事件的发生（比如RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了），如果我们需要对这种小概率事件也要管理起来，那么我们要用到事务。由于这里仅为RabbitMQ的简单介绍，所以这里将不讲解RabbitMQ相关的事务。

### Prefetch count ###

如果有多个消费者同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者。这时如果每个消息的处理时间不同，就有可能会导致某些消费者一直在忙，而另外一些消费者很快就处理完手头工作并一直空闲的情况。我们可以通过设置prefetchCount来限制Queue每次发送给每个消费者的消息数，比如我们设置prefetchCount=1，则Queue每次给每个消费者发送一条消息；消费者处理完这条消息后Queue会再给该消费者发送一条消息。

![20180417192935418][]

### Exchange ###

生产者将消息投递到Queue中，实际上在RabbitMQ中这种事情永远都不会发生。实际的情况是，生产者将消息发送到Exchange（交换器，下图中的X），由Exchange将消息路由到一个或多个Queue中(或者丢弃)

![20180420140708822][]

Exchange是按照什么逻辑将消息路由到Queue的？RabbitMQ中的Exchange有四种类型，不同的类型有着不同的路由策略，这将在Exchange Types一节介绍。

### routing key ###

生产者在将消息发送给Exchange的时候，一般会指定一个routing key，来指定这个消息的路由规则，而这个routing key需要与Exchange Type及binding key联合使用才能最终生效。 在Exchange Type与binding key固定的情况下（在正常使用时一般这些内容都是固定配置好的），我们的生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪里。 RabbitMQ为routing key设定的长度限制为255 bytes。

### Binding ###

RabbitMQ中通过Binding将Exchange与Queue关联起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。

![20180420172505586][]

### Binding key ###

在绑定（Binding）Exchange与Queue的同时，一般会指定一个binding key；消费者将消息发送给Exchange时，一般会指定一个routing key；当binding key与routing key相匹配时，消息将会被路由到对应的Queue中。这个将在Exchange Types章节会列举实际的例子加以说明。 在绑定多个Queue到同一个Exchange的时候，这些Binding允许使用相同的binding key。 binding key 并不是在所有情况下都生效，它依赖于Exchange Type，比如fanout类型的Exchange就会无视binding key，而是将消息路由到所有绑定到该Exchange的Queue。

### Exchange Types ###

RabbitMQ常用的Exchange Type有fanout、direct、topic、headers这四种（AMQP规范里还提到两种Exchange Type，分别为system与自定义，这里不予以描述），下面分别进行介绍。

### fanout ###

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

![20180420174502184][]

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

### direct ###

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

![20180420174637631][]

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

### topic ###

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中可以存在两种特殊字符“\*”与“\#”，用于做模糊匹配，其中“\*”用于匹配一个单词，“\#”用于匹配多个单词（可以是零个）

![20180420175344398][]

以上图中的配置为例，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都匹配)；

### headers ###

headers类型的Exchange不依赖于routing key与binding key的匹配规则来路由消息，而是根据发送的消息内容中的headers属性进行匹配。 在绑定Queue与Exchange时指定一组键值对；当消息发送到Exchange时，RabbitMQ会取到该消息的headers（也是一个键值对的形式），对比其中的键值对是否完全匹配Queue与Exchange绑定时指定的键值对；如果完全匹配则消息会路由到该Queue，否则不会路由到该Queue。 该类型的Exchange没有用到过（不过也应该很有用武之地），所以不做介绍。

\#\#\#\#\# 2.2.2 MQ数据丢失

> 1. 存储在队列中，如果队列没有对消息持久化，RabbitMQ服务器宕机重启会丢失数据。

解决：消息持久化需要设置，消息持久化可以防止消息在RabbitMQ Server中不会因为宕机重启而丢失，Exchange持久化和Queue持久化。

> 2. 生产者发送消息到RabbitMQ服务器过程中，RabbitMQ服务器如果宕机停止服务，消息会丢失。

解决：confirm机制，如果有异常将返回的消息存储到日志表中，再使用定时任务调度签收处理

> 3. 消费者从RabbitMQ服务器获取队列中存储的数据消费，但是消费者程序出错或者宕机而没有正确消费，导致数据丢失

解决：消费成功确认回复调用basicAck(消费成功)。如果抛出异常，回复basicNack

[点击打开链接][Link 1]

[70]: /images/20220527/68a8eeee54784713b61039e82c703153.png
[20180417184758767]: /images/20220527/2b212c9cf50f4d068538265db10b8e54.png
[20180417191719289]: /images/20220527/a87ae84d1a674673b05bb296230a866f.png
[20180417192935418]: /images/20220527/002f3810fbce4d5aa80254b793e3dc28.png
[20180420140708822]: /images/20220527/4a7e11afd9214539bf74fa60c7cafc9f.png
[20180420172505586]: /images/20220527/22ea88eab2f1415cbcbd38333e8abb5f.png
[20180420174502184]: /images/20220527/75e95e55842849768a2cecf491685419.png
[20180420174637631]: /images/20220527/d1dfcfed24de4fce898b6318a99b442d.png
[20180420175344398]: /images/20220527/9d70074d32fc464ab06af9f52400fcd6.png
[Link 1]: https://www.sojson.com/blog/48.html