AMQP协议详解与RabbitMQ，MQ消息队列的应用场景，如何避免消息丢失等消息队列常见问题

什么是AMQP?

  在异步通信中，消息不会马上到达接收方，而是被存放到一个容器中，当知足必定的条件以后，消息会被容器发送给接收方，这个容器即消息队列，而完成这个功能须要双方和容器以及其中的各个组件遵照统一的约定和规则，
AMQP就是这样的一种协议，消息发送与接受的双方遵照这个协议能够实现异步通信。这个协议约定了消息的格式和工做方式。

AMQP 中包含的主要元素

生产者（Producer）：向Exchange发布消息的应用。

消费者（Consumer）：从消息队列queue中消费消息的应用。

消息队列（Message Queue）：服务器组件，用于保存消息，直到发送给消费者。

Queue：消息载体；每一个消息都会被投入到一个或多个队列。

消息（Message）：传输的内容。

交换器（exchange）：路由组件，接收Producer发送的消息，并根据Routing Key转发给消息队列queue。

Routing Key：路由关键字，exchange根据这个Routing Key进行消息投递到队列queue。

 

虚拟主机（Virtual Host）: 用做不一样用户的权限分离;一批交换器，消息队列和相关对象。虚拟主机是共享相同身份认证和加密环境的独立服务器域。vhost 能够理解为虚拟 broker ，即 mini-RabbitMQ server。其内部均含有独立的 queue、exchange 和 binding 等，但最最重要的是，其拥有独立的权限系统，能够作到 vhost 范围的用户控制。固然，从 RabbitMQ 的全局角度，vhost 能够做为不一样权限隔离的手段（一个典型的例子就是不一样的应用能够跑在不一样的权限的 vhost 中）

Broker ：AMQP的服务端称为Broker。

链接（Connection）:一个网络链接，好比TCP/IP套接字链接；应用程序与Rabbit之间创建链接的管理器，程序代码中使用ConnectionFactory（链接管理器）。

信道（Channel）：消息通道，在客户端的每一个Connection链接里，可创建多个channel，每一个channel表明一个会话任务；多路复用链接中的一条独立的双向数据流通道，为会话提供物理传输介质。

绑定器（Binding）：把exchange和queue按照路由规则绑定起来。

 

exchange 与 Queue 的路由机制

生产者在发送消息时，都须要指定一个RoutingKey和Exchange，Exchange在接到该RoutingKey之后，会判断该ExchangeType，而后转发到对应的Queue中；

生产者发消息不须要指定Queue，消费者能够指定Queue绑定到某个RoutingKey和某个Exchange，也能够不指定Queue，就只根据某个Exchange和某个RoutingKey接受到消息。

 

Exchange 将消息发送到哪个queue是由exchange type 和 Binding绑定规则决定的，目前经常使用的有3种exchange，Direct exchange, Fanout exchange, Topic exchange :

　　1. Direct exchange 直接转发路由，其实现原理是会将消息中的RoutingKey与该Exchange关联的全部Binding中的BindingKey进行比较，若是相等，则发送到该Binding对应的Queue中。 

　　2. Fanout exchange 复制分发路由，该路由不须要RoutingKey，会将消息发送给全部与该 Exchange 定义过Binding的全部Queues中去，实际上是一种广播行为。 

　　3. topic exchange 通配路由，是direct exchange的通配符模式，消息中的RoutingKey能够写成通配的模式，exchange支持“#”和“*” 的通配。收到消息后，将消息转发给全部符合匹配正则表达式的Queue。 

 

TopicExchange的匹配符号：

 

#：匹配多个词

 

*： 匹配一个词

 

须要注意的一点只有queue具备保存消息的功能，exchange不能保存消息。

 

 

RabbitMQ中一个核心的原则是，消息不能直接投递到Queue中。Producer只能将本身的消息投递到Exchange中，由Exchange按照路由规则将消息投递到对应的Queue中。

 

在Consumer中，声明本身对哪一个Exchange感兴趣，并将本身的Queue绑定到本身感兴趣的路由关键字上，创建相应的映射关系；第二，在Producer中，将消息投递一个Exchange中，并指明它的路由关键字。

 

 

AMQP 如何实现通讯的

（1）创建链接Connection。由producer和consumer分别链接到broker的物理节点上。 

（2）创建消息Channel。Channel是创建在Connection之上的，一个Connection能够创建多个Channel；producer链接Virtual Host 创建Channel，Consumer链接到相应的queue上创建Channel。 

（3）发送消息。由Producer发送消息到Broker中的exchange中。 

（4）路由转发。exchange收到消息后，根据必定的路由策略routing key，将消息转发到相应的queue中去。 

（5）消息接收。Consumer会监听相应的queue，一旦queue中有能够消费的消息，queue就将消息发送给Consumer端。 

（6）消息确认。当Consumer完成某一条消息的处理以后，须要发送一条ACK消息给对应的Queue。Queue收到ACK信息后，才会认为消息处理成功，并将消息从Queue中移除；若是在对应的Channel断开后，Queue没有收到这条消息的ACK信息，该消息将被发送给另外的Channel。 至此一个消息的发送接收流程走完了。消息的确认机制提升了通讯的可靠性。

 

消息队列的使用大概过程

（1）客户端链接Connection到消息队列服务器Broker，打开一个channel。
（2）客户端声明一个exchange，并设置相关属性。
（3）客户端声明一个queue，并设置相关属性。
（4）客户端使用routing key，在exchange和queue之间创建好绑定关系。
（5）客户端投递消息到exchange。

RabbitMQ中 exchange、route、queue的关系

MessageQueue、Exchange和Binding构成了AMQP协议的核心。

　　声明MessageQueue 

 

　　在Rabbit MQ中，不管是生产者发送消息仍是消费者接受消息，都首先须要声明一个MessageQueue。这就存在一个问题，是生产者声明仍是消费者声明呢？要解决这个问题，首先须要明确：

 

　　a)消费者是没法订阅或者获取不存在的MessageQueue中信息。

 

　　b)消息被Exchange接受之后，若是没有匹配的Queue，则会被丢弃。

 

　　在明白了上述两点之后，就容易理解若是是消费者去声明Queue，就有可能会出如今声明Queue以前，生产者已发送的消息被丢弃的隐患。若是应用可以经过消息重发的机制容许消息丢失，则使用此方案没有任何问题。可是若是不能接受该方案，这就须要不管是生产者仍是消费者，在发送或者接受消息前，都须要去尝试创建消息队列。

 

　　

　　(重点) 这里有一点须要明确：

• 若是客户端尝试创建一个已经存在的消息队列，Rabbit MQ不会作任何事情，并返回客户端创建成功的，因此一个队列若是已经存在了，好比消费者若是再次尝试创建已存在的队列，是无效的
• 好比，你经过SpringBoot程序已经创建了一个queueA，再经过另外一个SpringBoot程序想要更改其queue属性，好比设置队列持久化durable=="true",就再次创建了一个queueA设置属性，是无效的

 

       若是一个消费者在一个信道中正在监听某一个队列的消息，Rabbit MQ是不容许该消费者在同一个channel去声明其余队列的。Rabbit MQ中，能够经过queue.declare命令声明一个队列，能够设置该队列如下属性:

 

a) Exclusive：排他队列，若是一个队列被声明为排他队列，该队列仅对首次声明它的链接可见，并在链接断开时自动删除。这里须要注意三点：其一，排他队列是基于链接可见的，同一链接的不一样信道是能够同时访问同一个链接建立的排他队列的。其二，“首次”，若是一个链接已经声明了一个排他队列，其余链接是不容许创建同名的排他队列的，这个与普通队列不一样。其三，即便该队列是持久化的，一旦链接关闭或者客户端退出，该排他队列都会被自动删除的。这种队列适用于只限于一个客户端发送读取消息的应用场景。

 

b)   Auto-delete:自动删除，若是该队列没有任何订阅的消费者的话，该队列会被自动删除。这种队列适用于临时队列。

 

 c)   Durable:持久化。

 

d)  其余选项，例如若是用户仅仅想查询某一个队列是否已存在，若是不存在，不想创建该队列，仍然能够调用queue.declare，只不过须要将参数passive设为true，传给queue.declare，若是该队列已存在，则会返回true；若是不存在，则会返回Error，可是不会建立新的队列。

 

　　

 

exchange 与 Queue 的路由机制

生产者在发送消息时，都须要指定一个RoutingKey和Exchange，Exchange在接到该RoutingKey之后，会判断该ExchangeType，而后转发到对应的Queue中，因此发消息不须要指定Queue，彷佛消费者能够指定Queue绑定到某个RoutingKey和某个Exchange，也能够不指定Queue，就只根据某个Exchange和某个RoutingKey接受到消息。

exchange 将消息发送到哪个queue是由exchange type 和 Binding绑定规则决定的，目前经常使用的有3种exchange，Direct exchange, Fanout exchange, Topic exchange 。 

Direct exchange 直接转发路由，其实现原理是会将消息中的RoutingKey与该Exchange关联的全部Binding中的BindingKey进行比较，若是相等，则发送到该Binding对应的Queue中。 

Fanout exchange 复制分发路由，该路由不须要RoutingKey，会将消息发送给全部与该 Exchange 定义过Binding的全部Queues中去，实际上是一种广播行为。 

topic exchange 通配路由，是direct exchange的通配符模式，消息中的RoutingKey能够写成通配的模式，exchange支持“#”和“*” 的通配。收到消息后，将消息转发给全部符合匹配正则表达式的Queue。 

须要注意的一点只有queue具备保存消息的功能，exchange不能保存消息。

AMQP的应用场景

AMQP是实现消息机制的一种协议，消息队列主要有如下几种应用场景：

异步处理

跨系统的异步通讯；好比公司新入职一个员工，须要开通系统帐号，有几件事情要作，开通系统帐号，发短信通知用户，发邮件给员工，在公司内部通信系统中发送消息给员工。其中发短信，发邮件，发内部通信系统消息，这三件事情能够串行也能够并行，并行的好处就是能够提升效率，这时能够应用MQ来实现并行；若是不使用MQ，那么开通系统帐号的服务就要依次调用发短信服务、发邮件服务、发内部通信系统消息，之后若是还要分配git帐号，那又要在开通系统帐号的服务里添加代码，若是使用MQ那么开通帐号后发送消息到MQ，对应订阅的消费者消费就好了。

异步处理不须要返回值的耗时操做：https://www.cnblogs.com/theRhyme/p/10796009.html

应用解耦

在公司内部系统中，有人事系统，OA系统，财务系统，外围应用系统等等，当人事发生变更的时候（离职入职调岗），人事系统须要将这些变更通知给其余系统，这时只需人事系统发送一条消息，各个外围系统订阅该消息，就可得知人事变更，与实时服务调用相比，若是人事系统挂掉，各个外围系统不会受到影响，继续运行；若是是实时服务调用，好比人事系统被各个服务调用，人事系统挂了，调用人事系统的服务都会受到影响。

 

死信队列

重要的业务队列若是挂了，能够被从新路由到死信队列进行处理。

 

分布式事务

RocketMQ TODO待写

流量缓冲

在有些流量会瞬间暴增的场景下，如秒杀，为了防止流量忽然增大而使得应用挂掉，能够引入MQ，将请求存入MQ中，若是超过了MQ的长度，就把请求丢弃掉，这样来限制流量。

日志处理

将消息队列引入到日志处理中，如kafka的应用，解决了大量日志的传输问题。日志客户端负责采集日志数据，并按期写入kafka队列，kafka负责接收，存储和转发日志，日志处理系统订阅并消费kafka中的日志数据。

 

SpringBoot+RabbitMQ的简单demo

 https://www.cnblogs.com/theRhyme/p/10071781.html

 

RabbitMQ死信队列的应用场景和代码实现

http://www.javashuo.com/article/p-njifrvnw-mn.html

 

RabbitMQ延迟队列代码实现和应用场景

场景： 订单下单30min若是没有付款就删除该订单

经过消息过时后进入死信交换器，再由交换器转发到延迟消费队列（重定向队列），实现延迟功能；

使用 rabbitmq_delayed_message_exchange 插件实现延迟功能。

代码：http://www.javashuo.com/article/p-rrimoagi-mo.html

 

rabbitmq 怎么避免消息丢失？

• 生产者Confirm机制（异步，推荐）或是事务方式（同步，不推荐）
• MQ服务端将消息持久化
• 消费者给MQ回复ACK，确认机制
• MQ服务端设置集群镜像模式
• 消费者消费消息补偿机制（如死信队列）

 

若是生产者弄丢了数据

 

RabbitMQ 生产者将数据发送到 RabbitMQ 的时候，可能数据在网络传输中搞丢了，这个时候 RabbitMQ 收不到消息，消息就丢了。

RabbitMQ 提供了两种方式来解决这个问题：

 

事务方式：在生产者发送消息以前，经过`channel.txSelect`开启一个事务，接着发送消息。

若是消息没有成功被 RabbitMQ 接收到，生产者会收到异常，此时就能够进行事务回滚`channel.txRollback`，而后从新发送。假如 RabbitMQ 收到了这个消息，就能够提交事务`channel.txCommit`。

可是这样一来，生产者的吞吐量和性能都会下降不少，如今通常不这么干。

 

另一种方式就是经过 Confirm 机制：这个 Confirm 模式是在生产者那里设置的，就是每次发消息的时候会分配一个惟一的 ID，而后 RabbitMQ服务端 收到以后会回传一个 ACK，告诉生产者这个消息 OK 了。

若是 RabbitMQ 没有处理到这个消息，那么就回调一个 Nack 的接口，这个时候生产者就能够重发。

 

事务机制和 Confirm 机制最大的不一样在于事务机制是同步的，提交一个事务以后会阻塞在那儿。 

可是 Confirm 机制是异步的，发送一个消息以后就能够发送下一个消息，而后那个消息 RabbitMQ 接收了以后会异步回调你一个接口通知你这个消息接收到了。

因此通常在生产者这块避免数据丢失，都是用 Confirm 机制的。

 

要保证消息持久化成功的条件有哪些？

• 声明队列必须设置持久化 durable 设置为 true.
• 消息推送投递模式必须设置持久化，deliveryMode 设置为 2（持久）。
• 消息已经到达持久化交换器。
• 消息已经到达持久化队列。

以上四个条件都知足才能保证消息持久化成功。

 

rabbitmq 持久化有什么缺点？

持久化的缺点就是下降了服务器的吞吐量，由于使用的是磁盘而非内存存储，从而下降了吞吐量。可尽可能使用 ssd 硬盘来缓解吞吐量的问题。

 

RabbitMQ如何保证同一个队列中的消息被顺序消费？

TODO待写

 

 

来源：

http://www.javashuo.com/article/p-zcdowiti-gw.html

http://www.javashuo.com/article/p-qalwxbqb-hh.html

http://www.javashuo.com/article/p-sahvupop-hq.html

http://techblog.ppdai.com/2018/07/17/20180717/

https://www.toutiao.com/a6698312611185820171/?timestamp=1559696015&app=news_article&group_id=6698312611185820171&tdsourcetag=s_pctim_aiomsg&req_id=2019060508533401002506701591332CD

https://mp.weixin.qq.com/s?__biz=MjM5ODI5Njc2MA==&mid=2655825391&idx=1&sn=f7523195ff08a51085012c736bc002a8&chksm=bd74e0388a03692e49ca3967a03dd2e8164e02741a75cabc8bd56e5b70ba6f2cc19f6fe10fd2&scene=0&xtrack=1&key=1c855a3d2871be72b53c28efecdb6c847aa5d9daffdac4207cc93d6a62948c3ac03b6e8813a35eaa72a54f7668de41b31fb1265ff3066312574ca210769ad2b726b9932dd21a296f9ea91fd6cf367dd7&ascene=1&uin=ODEzMzE3OTc%3D&devicetype=Windows+10&version=62060833&lang=zh_CN&pass_ticket=ZLGuBJ0cY2BIuQpqK%2Be08dQVFm3Htt7htVVelbWP8XE%3D