RabbitMQ技术详解（转）

时间 2019-11-12

原文原文链接

RabbitMQ是什么

定义

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

AMPQ

AMQP，即Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。
它可使对应的客户端（client）与对应的消息中间件（broker）进行交互。消息中间件从发布者（publisher）那里收到消息（发布消息的应用，也称为producer），而后将他们转发给消费者（consumers，处理消息的应用）。因为AMQP是一个网络协议，因此发布者、消费者以及消息中间件能够部署到不一样的物理机器上面。node

虽然在同步消息通信的世界里有不少公开标准（如 COBAR的 IIOP ，或者是 SOAP 等），可是在异步消息处理中却不是这样，只有大企业有一些商业实现（如微软的 MSMQ ，IBM 的 Websphere MQ 等），所以，在 2006 年的 6 月，Cisco 、Redhat、iMatix 等联合制定了 AMQP 的公开标准。mysql

RabbitMQ是由RabbitMQ Technologies Ltd开发而且提供商业支持的。该公司在2010年4月被SpringSource（VMWare的一个部门）收购。在2013年5月被并入Pivotal。其实VMWare，Pivotal和EMC本质上是一家的。不一样的是VMWare是独立上市子公司，而Pivotal是整合了EMC的某些资源，如今并无上市。git

总结

最终，上述同类产品：

都有各自客户端API或支持多种编程语言
都有大量的文档
都提供了积极的支持
ActiveMQ、RabbitMQ、MSMQ、Redis都须要启动服务进程，这些均可以监控和配置，其余几个就有问题了
都相对提供了良好的可靠性（一致性）、扩展性和负载均衡，固然还有性能

下面是网友的对四种消息队列的具体测试结果，显示的是发送和接受的每秒钟的消息数，整个过程共产生1百万条1K的消息，测试的执行是在WIndows Vista上进行的。（有待本身验证）

如上图所见，ZeroMQ和其余的不是一个级别，它的性能惊人的高。结论很清楚：若是但愿一个应用程序发送消息越快越好，选择ZeroMQ，而且在你不太在乎偶然会丢失某些消息的状况下更有价值。

RabbitMQ为什么会出现

或者说AMPQ为什么会出现，它的应用场景又是什么？

解决什么问题

你是否遇到过两个（多个）系统间须要经过定时任务来同步某些数据？你是否在为异构系统的不一样进程间相互调用、通信的问题而苦恼、挣扎？
在Web应用高并发环境下，因为来不及同步处理，请求每每会发生堵塞。好比说，大量的insert、update请求同时到达mysql，会带来无数的行锁表锁，最后致使请求数过多，触发too many connections错误。

消息服务擅长于解决多系统、异构系统间的数据交换（消息通知/通信）问题，你也能够把它用于系统间服务的相互调用（RPC）经过使用消息队列，咱们能够异步处理请求，从而缓解系统的压力。

应用场景

对于一个大型的软件系统来讲，它会有不少的组件或者说模块或者说子系统或者（Subsystem or Component or Submodule）。那么这些模块的如何通讯？这和传统的IPC有很大的区别。传统的IPC不少都是在单一系统上的，模块耦合性很大，不适合扩展（Scalability）；若是使用socket那么不一样的模块的确能够部署到不一样的机器上，可是仍是有不少问题须要解决。好比：

信息的发送者和接收者如何维持这个链接，若是一方的链接中断，这期间的数据如何方式丢失？
如何下降发送者和接收者的耦合度？
如何让Priority高的接收者先接到数据？
如何作到Load balance？有效均衡接收者的负载？
如何有效的将数据发送到相关的接收者？也就是说将接收者subscribe 不一样的数据，如何作有效的filter。
如何作到可扩展，甚至将这个通讯模块发到cluster上？
如何保证接收者接收到了完整，正确的数据？

AMDQ协议解决了以上的问题，而RabbitMQ实现了AMQP。

RabbitMQ基础概念

应用场景架构

RabbitMQ Server：也叫broker server，它不是运送食物的卡车，而是一种传输服务。原话是RabbitMQ isn’t a food truck, it’s a delivery service. 他的角色就是维护一条从Producer到Consumer的路线，保证数据可以按照指定的方式进行传输。可是这个保证也不是100%的保证，可是对于普通的应用来讲这已经足够了。固然对于商业系统来讲，能够再作一层数据一致性的guard，就能够完全保证系统的一致性了。
Client A & B：也叫Producer，数据的发送方。Create messages and Publish (Send) them to a broker server (RabbitMQ)。一个Message有两个部分：Payload（有效载荷）和Label（标签）。Payload顾名思义就是传输的数据，Label是Exchange的名字或者说是一个tag，它描述了payload，并且RabbitMQ也是经过这个label来决定把这个Message发给哪一个Consumer。AMQP仅仅描述了label，而RabbitMQ决定了如何使用这个label的规则。
Client 1，2，3：也叫Consumer，数据的接收方。Consumers attach to a broker server (RabbitMQ) and subscribe to a queue。把queue比做是一个有名字的邮箱。当有Message到达某个邮箱后，RabbitMQ把它发送给它的某个订阅者即Consumer。固然可能会把同一个Message发送给不少的Consumer。在这个Message中，只有payload，label已经被删掉了。对于Consumer来讲，它是不知道谁发送的这个信息的。就是协议自己不支持。可是固然了若是Producer发送的payload包含了Producer的信息就另当别论了。

对于一个数据从Producer到Consumer的正确传递，还有三个概念须要明确：exchanges, queues and bindings。

Exchanges are where producers publish their messages. 消息交换机，它指定消息按什么规则，路由到哪一个队列
Queues are where the messages end up and are received by consumers. 消息队列载体，每一个消息都会被投入到一个或多个队列
Bindings are how the messages get routed from the exchange to particular queues. 绑定，它的做用就是把exchange和queue按照路由规则绑定起来
Routing Key：路由关键字，exchange根据这个关键字进行消息投递

还有几个概念是上述图中没有标明的，那就是Connection（链接），Channel（通道，频道），Vhost（虚拟主机）。

Connection：就是一个TCP的链接。Producer和Consumer都是经过TCP链接到RabbitMQ Server的。之后咱们能够看到，程序的起始处就是创建这个TCP链接。
Channel：虚拟链接。它创建在上述的TCP链接中。数据流动都是在Channel中进行的。也就是说，通常状况是程序起始创建TCP链接，第二步就是创建这个Channel。
Vhost：虚拟主机，一个broker里能够开设多个vhost，用做不一样用户的权限分离。每一个virtual host本质上都是一个RabbitMQ Server，拥有它本身的queue，exchagne，和bings rule等等。这保证了你能够在多个不一样的application中使用RabbitMQ。

Channel的选择

那么，为何使用Channel，而不是直接使用TCP链接？

对于OS来讲，创建和关闭TCP链接是有代价的，频繁的创建关闭TCP链接对于系统的性能有很大的影响，并且TCP的链接数也有限制，这也限制了系统处理高并发的能力。可是，在TCP链接中创建Channel是没有上述代价的。对于Producer或者Consumer来讲，能够并发的使用多个Channel进行Publish或者Receive。
有实验代表，1s的数据能够Publish10K的数据包。固然对于不一样的硬件环境，不一样的数据包大小这个数据确定不同，可是我只想说明，对于普通的Consumer或者Producer来讲，这已经足够了。若是不够用，你考虑的应该是如何细化split你的设计。

消息队列执行过程

客户端链接到消息队列服务器，打开一个Channel。
客户端声明一个Exchange，并设置相关属性。
客户端声明一个Queue，并设置相关属性。
客户端使用Routing key，在Exchange和Queue之间创建好绑定关系。
客户端投递消息到Exchange。

Exchange接收到消息后，就根据消息的key和已经设置的Binding，进行消息路由，将消息投递到一个或多个队列里。有三种类型的Exchanges：direct，fanout，topic，每一个实现了不一样的路由算法（routing algorithm）：

Direct exchange：彻底根据key进行投递的叫作Direct交换机。若是Routing key匹配, 那么Message就会被传递到相应的queue中。其实在queue建立时，它会自动的以queue的名字做为routing key来绑定那个exchange。例如，绑定时设置了Routing key为”abc”，那么客户端提交的消息，只有设置了key为”abc”的才会投递到队列。
Fanout exchange：不须要key的叫作Fanout交换机。它采起广播模式，一个消息进来时，投递到与该交换机绑定的全部队列。
Topic exchange：对key进行模式匹配后进行投递的叫作Topic交换机。好比符号”#”匹配一个或多个词，符号””匹配正好一个词。例如”abc.#”匹配”abc.def.ghi”，”abc.”只匹配”abc.def”。

更多消息队列相关设计介绍请参考：

消息队列的建立

Consumer和Procuder均可以经过 queue.declare 建立queue。对于某个Channel来讲，Consumer不能declare一个queue，却订阅其余的queue。固然也能够建立私有的queue。这样只有app自己才可使用这个queue。queue也能够自动删除，被标为auto-delete的queue在最后一个Consumer unsubscribe后就会被自动删除。那么若是是建立一个已经存在的queue呢？那么不会有任何的影响。须要注意的是没有任何的影响，也就是说第二次建立若是参数和第一次不同，那么该操做虽然成功，可是queue的属性并不会被修改。

那么谁应该负责建立这个queue呢？是Consumer，仍是Producer？

若是queue不存在，固然Consumer不会获得任何的Message。可是若是queue不存在，那么Producer Publish的Message会被丢弃。因此，仍是为了数据不丢失，Consumer和Producer都try to create the queue！反正无论怎么样，这个接口都不会出问题。

Queue对load balance的处理是完美的。对于多个Consumer来讲，RabbitMQ 使用循环的方式（round-robin）的方式均衡的发送给不一样的Consumer。

消息的ack机制

默认状况下，若是Message 已经被某个Consumer正确的接收到了，那么该Message就会被从queue中移除。固然也可让同一个Message发送到不少的Consumer。

若是一个queue没被任何的Consumer Subscribe（订阅），那么，若是这个queue有数据到达，那么这个数据会被cache，不会被丢弃。当有Consumer时，这个数据会被当即发送到这个Consumer，这个数据被Consumer正确收到时，这个数据就被从queue中删除。

那么什么是正确收到呢？经过ack。

每一个Message都要被acknowledged（确认，ack）。咱们能够显示的在程序中去ack（Consumer的basic.ack），也能够自动的ack（订阅Queue时指定auto_ack为true）。
若是有数据没有被ack，那么RabbitMQ Server会把这个信息发送到下一个Consumer。
若是这个app有bug，忘记了ack，那么RabbitMQ Server不会再发送数据给它，由于Server认为这个Consumer处理能力有限。
并且ack的机制能够起到限流的做用（Benefit to throttling）：在Consumer处理完成数据后发送ack，甚至在额外的延时后发送ack，将有效的balance Consumer的load。

固然对于实际的例子，好比咱们可能会对某些数据进行merge，好比merge 4s内的数据，而后sleep 4s后再获取数据。特别是在监听系统的state，咱们不但愿全部的state实时的传递上去，而是但愿有必定的延时。这样能够减小某些IO，并且终端用户也不会感受到。

没有正确响应呢？

若是Consumer接收了一个消息就尚未发送ack就与RabbitMQ断开了，RabbitMQ会认为这条消息没有投递成功会从新投递到别的Consumer。
若是Consumer自己逻辑有问题没有发送ack的处理，RabbitMQ不会再向该Consumer发送消息。RabbitMQ会认为这个Consumer尚未处理完上一条消息，没有能力继续接收新消息。

咱们能够善加利用这一机制，若是须要处理过程是至关复杂的，应用程序能够延迟发送ack直处处理完成为止。这能够有效控制应用程序这边的负载，不致于被大量消息冲击。

消息拒绝

因为要拒绝消息，因此ack响应消息尚未发出，这里拒绝消息能够有两种选择:

Consumer直接断开RabbitMQ这样RabbitMQ将把这条消息从新排队，交由其它Consumer处理。这个方法在RabbitMQ各版本都支持，这样作的坏处就是链接断开增长了RabbitMQ的额外负担，特别是consumer出现异常每条消息都没法正常处理的时候。
RabbitMQ 2.0.0可使用 basic.reject 命令，收到该命令RabbitMQ会从新投递到其它的Consumer。若是设置requeue为false，RabbitMQ会直接将消息从queue中移除。
其实还有一种选择就是直接忽略这条消息并发送ACK，当你明确知道这条消息是异常的不会有Consumer能处理，能够这样作抛弃异常数据。

为何要发送basic.reject消息而不是ACK？RabbitMQ后面的版本可能会引入”dead letter”队列，若是想利用dead letter作点文章就使用basic.reject并设置requeue为false。

消息持久化

RabbitMQ支持消息的持久化，也就是数据写在磁盘上，为了数据安全考虑，大多数用户都会选择持久化。消息队列持久化包括3个部分：

Exchange持久化，在声明时指定durable => 1
Queue持久化，在声明时指定durable => 1
消息持久化，在投递时指定delivery_mode => 2（1是非持久化）

若Exchange和Queue都是持久化的，那么它们之间的Binding也是持久化的；而Exchange和Queue二者之间有一个持久化，一个非持久化，就不容许创建绑定。

Consumer从durable queue中取回一条消息以后并发回了ack消息，RabbitMQ就会将其标记，方便后续垃圾回收。若是一条持久化的消息没有被consumer取走，RabbitMQ重启以后会自动重建exchange和queue(以及bingding关系)，消息经过持久化日志重建再次进入对应的queues，exchanges。

RabbitMQ集群

因为RabbitMQ是用erlang开发的，RabbitMQ彻底依赖erlang的Cluster，由于erlang天生就是一门分布式语言，集群很是方便，但其自己并不支持负载均衡。Erlang的集群中各节点是经由过程一个magic cookie来实现的，这个cookie存放在 $home/.erlang.cookie中(像个人root用户安装的就是放在个人root/.erlang.cookie中)，文件是400的权限。因此必须保证各节点cookie内容一致，否则节点之间就没法通讯。

集群方式

Rabbitmq集群大概分为二种方式：

普通模式：默认的集群模式。
对于Queue来讲，消息实体只存在于其中一个节点，A、B两个节点仅有相同的元数据，即队列结构，但队列的元数据仅保存有一份，即建立该队列的rabbitmq节点（A节点），当A节点宕机，你能够去其B节点查看，./rabbitmqctl list_queues 发现该队列已经丢失，但声明的exchange还存在。

当消息进入A节点的Queue中后，consumer从B节点拉取时，RabbitMQ会临时在A、B间进行消息传输，把A中的消息实体取出并通过B发送给consumer，因此consumer应平均链接每个节点，从中取消息。

该模式存在一个问题就是当A节点故障后，B节点没法取到A节点中还未消费的消息实体。若是作了队列持久化或消息持久化，那么得等A节点恢复，而后才可被消费，而且在A节点恢复以前其它节点不能再建立A节点已经建立过的持久队列；若是没有持久化的话，消息就会失丢。

这种模式更适合非持久化队列，只有该队列是非持久的，客户端才能从新链接到集群里的其余节点，并从新建立队列。假如该队列是持久化的，那么惟一办法是将故障节点恢复起来。
镜像模式：把须要的队列作成镜像队列，存在于多个节点。

该模式解决了普通模式的问题，其实质不一样之处在于，消息实体会主动在镜像节点间同步，而不是在consumer取数据时临时拉取。

该模式带来的反作用也很明显，除了下降系统性能外，若是镜像队列数量过多，加之大量的消息进入，集群内部的网络带宽将会被这种同步通信大大消耗掉。

因此在对可靠性要求较高的场合中适用，一个队列想作成镜像队列，须要先设置policy，而后客户端建立队列的时候，rabbitmq集群根据“队列名称”自动设置是普通集群模式或镜像队列。具体以下：

队列经过策略来使能镜像。策略能在任什么时候刻改变，rabbitmq队列也近可能的将队列随着策略变化而变化；非镜像队列和镜像队列之间是有区别的，前者缺少额外的镜像基础设施，没有任何slave，所以会运行得更快。

为了使队列称为镜像队列，你将会建立一个策略来匹配队列，设置策略有两个键“ha-mode和 ha-params（可选）”。ha-params根据ha-mode设置不一样的值，下面表格说明这些key的选项。

为何RabbitMQ不将队列复制到集群里每一个节点呢？这与它的集群的设计本意相冲突，集群的设计目的就是增长更多节点时，能线性的增长性能（CPU、内存）和容量（内存、磁盘）。理由以下：

Storage Space: If every cluster node had a full copy of every queue, adding nodes wouldn’t give you more storage capacity. For example, if one node could store 1GB of messages, adding two more nodes would simply give you two more copies of the same 1GB of messages.（存储空间：若是每一个集群节点每一个队列的一个完整副本，增长节点须要更多的存储容量。例如，若是一个节点能够存储1 gb的消息，添加两个节点须要两份相同的1gb的消息）
Performance: Publishing messages would require replicating those messages to every cluster node. For durable messages that would require triggering disk activity on all nodes for every message. Your network and disk load would increase every time you added a node, keeping the performance of the cluster the same (or possibly worse).（性能：发布消息须要将这些信息复制到每一个集群节点。对持久消息，要求为每条消息触发磁盘活动在全部节点上。每次添加一个节点都会带来网络和磁盘的负载。）

固然RabbitMQ新版本集群也支持队列复制（有个选项能够配置）。好比在有五个节点的集群里，能够指定某个队列的内容在2个节点上进行存储，从而在性能与高可用性之间取得一个平衡（应该就是指镜像模式）。

集群节点

RabbitMQ的集群节点包括内存节点、磁盘节点。顾名思义内存节点就是将全部数据放在内存，磁盘节点将数据放在磁盘。不过，若是在投递消息时，打开了消息的持久化，那么即便是内存节点，数据仍是安全的放在磁盘。

一个rabbitmq集群中能够共享 user，vhost，queue，exchange等，全部的数据和状态都是必须在全部节点上复制的，一个例外是，那些当前只属于建立它的节点的消息队列，尽管它们可见且可被全部节点读取。rabbitmq节点能够动态的加入到集群中，一个节点它能够加入到集群中，也能够从集群环集群会进行一个基本的负载均衡。

集群中有两种节点：