©2020-2024 ioDraw All rights reserved,
Privacy Policy
RabbitMQ：我们通常谈到消息队列，就会联想到这其中的三者：生产者、消费者和消息队列，生产者将消息发送到消息队列，消费者从消息队列中获取消息进行处理。对于RabbitMQ，它在此基础上做了一层抽象，引入了交换器exchange的概念，交换器是作用于生产者和消息队列之间的中间桥梁，它起了一种消息路由的作用，也就是说生产者并不和消息队列直接关联，而是先发送给交换器，再由交换器路由到对应的队列，至于它是根据何种规则路由到消息队列的，就是我们下面需要介绍的内容了。这里的生产者并没有直接将消息发送给消息队列，而是通过建立与Exchange（交换器）的Channel（信道），将消息发送给Exchange，Exchange根据路由规则，将消息转发给指定的消息队列。消息队列储存消息，等待消费者取出消息，消费者通过建立与消息队列相连的Channel,从消息队列中获取消息。
AMQP：Advanced Message
Queue，高级消息队列协议。它是应用层协议的一个开放标准，为面向消息的中间件设计，基于此协议的客户端与消息中间件可传递消息，并不受产品、开发语言等条件的限制。
实现原理图：
基本概念：
1.Channel（信道）：多路复用连接中的一条独立的双向数据流通道。信道是建立在真实的TCP连接内的虚拟连接，复用TCP连接的通道。
2.Producer（消息的生产者）：向消息队列发布消息的客户端应用程序。
3.Consumer（消息的消费者）：从消息队列取得消息的客户端应用程序。
4.Message（消息）：消息由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（消息优先权）、delivery-mode（是否持久性存储）等。
5.Routing Key（路由键）：消息头的一个属性，用于标记消息的路由规则，决定了交换机的转发路径。最大长度255 字节。
6.Queue（消息队列）：存储消息的一种数据结构，用来保存消息，直到消息发送给消费者。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将消息取走。需要注意，当多个消费者订阅同一个Queue，这时Queue中的消息会被平均分摊给多个消费者进行处理，而不是每个消费者都收到所有的消息并处理，每一条消息只能被一个订阅者接收。
7.Exchange（交换器|路由器）：提供Producer到Queue之间的匹配，接收生产者发送的消息并将这些消息按照路由规则转发到消息队列。交换器用于转发消息，它不会存储消息
，如果没有 Queue绑定到 Exchange 的话，它会直接丢弃掉 Producer
发送过来的消息。交换器有四种消息调度策略（下面会介绍），分别是fanout, direct, topic, headers。
8.Binding（绑定）：用于建立Exchange和Queue之间的关联。一个绑定就是基于Binding
Key将Exchange和Queue连接起来的路由规则，所以可以将交换器理解成一个由Binding构成的路由表。
6.Binding Key（绑定键）：Exchange与Queue的绑定关系，用于匹配Routing Key。最大长度255 字节。
7.Broker：RabbitMQ Server，服务器实体。
调度策略是指Exchange在收到生产者发送的消息后依据什么规则把消息转发到一个或多个队列中保存。调度策略与三个因素相关：Exchange
Type（Exchange的类型），Binding Key（Exchange和Queue的绑定关系），消息的标记信息（Routing
Key和headers）。Exchange根据消息的Routing Key和Exchange绑定Queue的Binding
Key分配消息。生产者将消息发送给Exchange时，一般会指定一个Routing Key，来指定这个消息的路由规则，而这个Routing
Key需要与Exchange Type及Binding Key联合使用才能最终生效。在Exchange Type与Binding
Key固定的情况下（一般这些内容都是固定配置好的），我们的生产者就可以在发送消息给Exchange时，通过指定Routing Key来决定消息流向哪里。
Exchange类型介绍：
1，Fanout（订阅模式|广播模式）
Fanout交换器会把所有发送到该交换器的消息路由到所有与该交换器绑定的消息队列中。订阅模式与Binding Key和Routing
Key无关，交换器将接受到的消息分发给有绑定关系的所有消息队列队列（不论Binding Key和Routing
Key是什么）。类似于子网广播，子网内的每台主机都获得了一份复制的消息。Fanout交换机转发消息是最快的。
2，Direct（路由模式）
Direct交换器需要消息的Routing Key与 Exchange和Queue 之间的Binding
Key完全匹配，如果匹配成功，将消息分发到该Queue。只有当Routing Key和Binding
Key完全匹配的时候，消息队列才可以获取消息。Direct是Exchange的默认模式。RabbitMQ默认提供了一个Exchange，名字是空字符串，类型是Direct，绑定到所有的Queue（每一个Queue和这个无名Exchange之间的Binding
Key是Queue的名字）。所以，有时候我们感觉不需要交换器也可以发送和接收消息，但是实际上是使用了RabbitMQ默认提供的Exchange。
3，Topic（通配符模式）
Topic交换器按照正则表达式模糊匹配：用消息的Routing Key与 Exchange和Queue 之间的Binding
Key进行模糊匹配，如果匹配成功，将消息分发到该Queue。Routing Key是一个句点号“. ”分隔的字符串（我们将被句点号“.
”分隔开的每一段独立的字符串称为一个单词）。Binding Key与Routing Key一样也是句点号“. ”分隔的字符串。Binding
Key中可以存在两种特殊字符“ * ”与“#”，用于做模糊匹配，其中“*”用于匹配一个单词，“#”用于匹配多个单词（也可以是零个或一个）。
例如下面这个消息队列的场景，用的是topic类型交换器：
RPC机制实现：
MQ本身是基于异步的消息处理，前面的示例中所有的生产者（P）将消息发送到RabbitMQ后不会知道消费者（C）处理成功或者失败，甚至连有没有消费者来处理这条消息都不知道。但实际的应用场景中，我们很可能需要一些同步处理，需要同步等待服务端将我的消息处理完成后再进行下一步处理。这相当于RPC（Remote
Procedure Call，远程过程调用）。
RabbitMQ中实现RPC的机制是：
1，生产者发送请求（消息）时，在消息的属性（MessageProperties，在AMQP协议中定义了14个属性，这些属性会随着消息一起发送）中设置两个属性值replyTo（一个Queue名称，用于告诉消费者处理完成后将通知我的消息发送到这个Queue中）和correlationId（此次请求的标识号，消费者处理完成后需要将此属性返还，生产者将根据这个id了解哪条请求被成功执行了或执行失败）。
2，消费者收到消息并处理。
3，消费者处理完消息后，将生成一条应答消息到replyTo指定的Queue，同时带上correlationId属性。
4，生产者之前已订阅replyTo指定的Queue，从中收到服务器的应答消息后，根据其中的correlationId属性分析哪条请求被执行了，根据执行结果进行后续业务处理。
消息确认机制：
在实际应用中，可能会发生消费者收到Queue中的消息，但没有处理完成就宕机（或出现其他意外）的情况，这种情况下就可能会导致消息丢失。为了避免这种情况发生，我们可以要求消费者在消费完消息后发送一个回执给RabbitMQ，RabbitMQ收到消息回执（Message
acknowledgment）后才将该消息从Queue中移除；如果RabbitMQ没有收到回执并检测到消费者的RabbitMQ连接断开，则RabbitMQ会将该消息发送给其他消费者（如果存在多个消费者）进行处理。这里不存在Timeout概念，一个消费者处理消息时间再长也不会导致该消息被发送给其他消费者，除非它的RabbitMQ连接断开。这里会产生另外一个问题，如果我们的开发人员在处理完业务逻辑后，忘记发送回执给RabbitMQ，这将会导致严重的问题，Queue中堆积的消息会越来越多，消费者重启后会重复消费这些消息并重复执行业务逻辑。如果我们采用no-ack的方式进行确认，也就是说，每次Consumer接到数据后，而不管是否处理完成，RabbitMQ会立即把这个Message标记为完成，然后从queue中删除了。
消息持久化机制：
如果我们希望即使在RabbitMQ服务重启的情况下，也不会丢失消息，我们可以将Queue与Message都设置为可持久化的（durable），这样可以保证绝大部分情况下我们的RabbitMQ消息不会丢失。但依然解决不了小概率丢失事件的发生（比如RabbitMQ服务器已经接收到生产者的消息，但还没来得及持久化该消息时RabbitMQ服务器就断电了），如果我们需要对这种小概率事件也要管理起来，那么我们要用到事务。
事务：
对事务的支持是AMQP协议的一个重要特性。假设当生产者将一个持久化消息发送给服务器时，因为consume命令本身没有任何Response返回，所以即使服务器崩溃，没有持久化该消息，生产者也无法获知该消息已经丢失。如果此时使用事务，即通过txSelect()开启一个事务，然后发送消息给服务器，然后通过txCommit()提交该事务，即可以保证，如果txCommit()提交了，则该消息一定会持久化，如果txCommit()还未提交即服务器崩溃，则该消息不会服务器接收。当然Rabbit
MQ也提供了txRollback()命令用于回滚某一个事务。
消息分发机制：
我们在应用程序使用消息系统时，一般情况下生产者往队列里插入数据时速度是比较快的，但是消费者消费数据往往涉及到一些业务逻辑处理导致速度跟不上生产者生产数据。因此如果一个生产者对应一个消费者的话，很容易导致很多消息堆积在队列里。这时，就得使用工作队列了。一个队列有多个消费者同时消费数据。工作队列有两种分发数据的方式：轮询分发（Round-robin）和
公平分发（Fair
dispatch）。轮询分发：队列给每一个消费者发送数量一样的数据。公平分发：消费者设置每次从队列中取一条数据，并且消费完后手动应答，继续从队列取下一个数据。
1，轮询分发
如果工作队列中有两个消费者，两个消费者得到的数据量一样的，并不会因为两个消费者处理数据速度不一样使得两个消费者取得不一样数量的数据。但是这种分发方式存在着一些隐患，消费者虽然得到了消息，但是如果消费者没能成功处理业务逻辑，在RabbitMQ中也不存在这条消息。就会出现消息丢失并且业务逻辑没能成功处理的情况。
2，公平分发
消费者设置每次从队列里取一条数据，并且关闭自动回复机制，每次取完一条数据后，手动回复并继续取下一条数据。与轮询分发不同的是，当每个消费都设置了每次只会从队列取一条数据时，并且关闭自动应答，在每次处理完数据后手动给队列发送确认收到数据。这样队列就会公平给每个消息费者发送数据，消费一条再发第二条，而且可以在管理界面中看到数据是一条条随着消费者消费完从而减少的，并不是一下子全部分发完了。采用公平分发方式就不会出现消息丢失并且业务逻辑没能成功处理的情况。