RabbitMQ基本概念

深碍√TFBOYSˉ_ 2022-06-02 05:16 252阅读 0赞

\#RabbitMQ架构

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70]  
说是架构其实更像是应用场景下的架构（自己画的有点丑，勿嫌弃）  
![这里写图片描述][format_png]  
RabbitMQ应用架构

从图中可以看出RabbitMQ主要由Exchange和Queue两部分组成，然后通过RoutingKey关联起来，消息投递到Exchange然后通过Queue接收。

## RabbitMQ消息队列基本概念 ##

**RabbitMQ Server**： 也叫broker server，它是一种传输服务。 他的角色就是维护一条从Producer到Consumer的路线，保证数据能够按照指定的方式进行传输。但是这个保证也不是100%的保证，但是对于普通的应用来说这已经足够了。当然对于商业系统来说，可以再做一层数据一致性的guard，就可以彻底保证系统的一致性了。

**Producer**： 消息生产者，如图A、B、C，数据的发送方。消息生产者连接RabbitMQ服务器然后将消息投递到Exchange。

**Consumer**：消息消费者，如图1、2、3，数据的接收方。消息消费者订阅队列，RabbitMQ将Queue中的消息发送到消息消费者。

**Exchange交换器**：生产者将消息发送到Exchange（交换器），由Exchange将消息路由到一个或多个Queue中（或者丢弃）。Exchange并不存储消息。RabbitMQ中的Exchange有fanout、direct、topic、headers四种类型，每种类型对应不同的路由规则，后面详细介绍这四种类型。  
![\[外链图片转存失败(img-nsWJPyyC-1569213336638)(http://ostest.qiniudn.com/wordpress/wp-content/uploads/2014/02/2014-2-21-9-51-03.png)\]][img-nsWJPyyC-1569213336638_http_ostest.qiniudn.com_wordpress_wp-content_uploads_2014_02_2014-2-21-9-51-03.png]

**Queue队列**：（队列）是RabbitMQ的内部对象，用于存储消息。消息消费者就是通过订阅队列来获取消息的，RabbitMQ中的消息都只能存储在Queue中，生产者生产消息并最终投递到Queue中，消费者可以从Queue中获取消息并消费。  
![\[外链图片转存失败(img-kSsZitXx-1569213336639)(http://ostest.qiniudn.com/wordpress/wp-content/uploads/2014/02/2014-2-21-9-46-43.png)\]][img-kSsZitXx-1569213336639_http_ostest.qiniudn.com_wordpress_wp-content_uploads_2014_02_2014-2-21-9-46-43.png]  
上图P到对队列之间是有一个Exchange（不指定的话系统默认有个defalut Exchange）  
多个消费者可以订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理。

**Prefetch count负载均衡**  
前面我们讲到如果有多个消费者同时订阅同一个Queue中的消息，Queue中的消息会被平摊给多个消费者。这时如果每个消息的处理时间不同，就有可能会导致某些消费者一直在忙，而另外一些消费者很快就处理完手头工作并一直空闲的情况。我们可以通过设置prefetchCount来限制Queue每次发送给每个消费者的消息数，比如我们设置prefetchCount=1，则Queue每次给每个消费者发送一条消息；消费者处理完这条消息后Queue会再给该消费者发送一条消息。  
![\[外链图片转存失败(img-4j6zehe1-1569213336639)(http://ostest.qiniudn.com/wordpress/wp-content/uploads/2014/02/2014-2-21-9-49-08.png)\]][img-4j6zehe1-1569213336639_http_ostest.qiniudn.com_wordpress_wp-content_uploads_2014_02_2014-2-21-9-49-08.png]

**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相关的事务。

**RoutingKey**：  
相当于Exchange的身份标识（唯一的标识），用以区分不同的Exchange.  
生产者在将消息发送给Exchange的时候，一般会指定一个routing key，用来标识这个路由的唯一标识，而这个routing key需要与Exchange Type(路由规则)及binding key(队列名称)联合使用才能最终生效。在Exchange Type与binding key固定的情况后，我们的生产者就可以在发送消息给Exchange时，通过指定routing key来决定消息流向哪个路由。RabbitMQ为routing key设定的长度限制为255 bytes。

如本文开头的那张图：Exchange1和Exchange2两个路由（至少有一个以上，一个时是系统默认的default Exchange），那么在发消息时就需要根据RoutingKey来确定发给谁，RoutingKey的值就是Exchange1和Exchange2

**Binding**  
RabbitMQ中通过Binding将Exchange与Queue关联起来，这样RabbitMQ就知道如何正确地将消息路由到指定的Queue了。  
因为一个Exchange可以绑定多个（Binding）Queue  
![在这里插入图片描述][20190923124801914.png]

**Binding key**  
相当于queue的身份标识（唯一的标识），用以区分queue.  
在绑定（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。

**Connection**： （连接）。Producer和Consumer都是通过TCP连接到RabbitMQ Server的。以后我们可以看到，程序的起始处就是建立这个TCP连接。

**Channels**： （信道）。它建立在上述的TCP连接中。数据流动都是在Channel中进行的。也就是说，一般情况是程序起始建立TCP连接，第二步就是建立这个Channel。

## Exchange Types ##

**direct一对一消息**  
默认的简单类型  
1.一般情况可以使用rabbitMQ自带的Exchange：该Exchange的名字为空字符串，下文称其为default Exchange  
2.这种模式下不需要将Exchange进行任何绑定(binding)操作  
3.消息传递时需要一个“RouteKey”，可以简单的理解为要发送到的队列名字。  
4.如果vhost中不存在RouteKey中指定的队列名，则该消息会被抛弃。  
![在这里插入图片描述][20190923125011734.png]

简述：只要发送一个RouteKey的字符串消息，那么如果存在RouteKey相同的队列那么该队列就会收到消息，否则该消息就被丢弃

**fanout广播消息**

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

*  这种模式不需要RouteKey
 *  这种模式需要提前将Exchange与Queue进行绑定，一个Exchange可以绑定多个Queue，一个Queue可以同多个Exchange进行绑定。
 *  如果接受到消息的Exchange没有与任何Queue绑定，则消息会被抛弃。
 *  ![在这里插入图片描述][20190923125118543.png]

这种类型，如同广播消息，只要exchange相同，则exchange绑定的queues都会收到相关消息

**topic订阅消息**

任何发送到Topic Exchange的消息都会被转发到所有关心RouteKey中指定话题的Queue上。  
![在这里插入图片描述][20190923125027502.png]  
前面讲到direct类型的Exchange路由规则是完全匹配binding key与routing key，但这种严格的匹配方式在很多情况下不能满足实际业务需求。topic类型的Exchange在匹配规则上进行了扩展，它与direct类型的Exchage相似，也是将消息路由到binding key与routing key相匹配的Queue中，但direct是完全匹配，而通过topic可以进行模糊匹配

*  这种模式Queue需要绑定RouteKey，Exchange要提前绑定一个或者多个这样的Queue。
 *  在进行绑定时，要提供一个该队列关心的主题topic绑定主题名称，如“\#.log.\#”表示该队列关心所有涉及log的消息(一个RouteKey为”MQ.log.error”的消息会被转发到该队列)。
 *  同direct，如果Exchange没有发现能够与RouteKey匹配的Queue，则会抛弃此消息。

> routing key为一个句点号“. ”分隔的字符串（我们将被句点号“.”分隔开的每一段独立的字符串称为一个单词），如“stock.usd.nyse”、“nyse.vmw”、“quick.orange.rabbit” binding key与routing key一样也是句点号“. ”分隔的字符串

*  “\#”表示0个或若干个关键字，
 *  “\*”表示一个关键字

如“log.\*”能与“log.warn”匹配，无法与“log.warn.timeout”匹配；但是“log.\#”能与上述两者匹配。

> binding key中可以存在两种特殊字符“*”与“\#”，“\#”表示0个或若干个关键字，“*”表示一个关键字。如“log.\*”能与“log.warn”匹配，无法与“log.warn.timeout”匹配；但是“log.\#”能与上述两者匹配。

![这里写图片描述][format_png 1]

**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_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70 1]  
RabbitMQ中实现RPC的机制是：

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

## 原文出处 ##

https://www.cnblogs.com/williamjie/p/9481774.html

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70]: /images/20220602/e1ec6bfc400a4185b21159402d3fc8a5.png
[format_png]: /images/20220602/204f664a6b3b43daa1c06577278ece6e.png
[img-nsWJPyyC-1569213336638_http_ostest.qiniudn.com_wordpress_wp-content_uploads_2014_02_2014-2-21-9-51-03.png]: /images/20220602/6dbf453d17a54c17b495fcb3bc5c955d.png
[img-kSsZitXx-1569213336639_http_ostest.qiniudn.com_wordpress_wp-content_uploads_2014_02_2014-2-21-9-46-43.png]: /images/20220602/b6a85a26607e4bb699d7bf5b2e99b39d.png
[img-4j6zehe1-1569213336639_http_ostest.qiniudn.com_wordpress_wp-content_uploads_2014_02_2014-2-21-9-49-08.png]: /images/20220602/e3a364544c8d4f69916fe272d9831341.png
[20190923124801914.png]: /images/20220602/49ec0cb695f246a89fc5b0ed916fb9a9.png
[20190923125011734.png]: /images/20220602/a1f6a21647b24a1fa350bc4708122463.png
[20190923125118543.png]: /images/20220602/3f8aa1afc8874ac8a0e651df62ec08bd.png
[20190923125027502.png]: /images/20220602/65fc5c3501344ccb99a771b473d7e8e8.png
[format_png 1]: /images/20220602/636f7298d2824b2ab0e8a83c19bfda58.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3pqY2phdmE_size_16_color_FFFFFF_t_70 1]: /images/20220602/891963dd06ad41489078c69e45390580.png