RabbitMQ演讲稿（精简版）

好几个月没有好好摸过RabbitMQ了，近日为了两场演讲花了点时间又重新整理了一番，下面给出精简过后的一个版本的大致内容。本来RabbitMQ的相关中文资料并不多，希望可以抛砖引玉，有更多的小伙伴可以分享下相关的经验。

（内容较多，光字就超过1W，建议先马后看）

消息中间件无非就是发送-存储-消费这三个方面，RabbitMQ也并不例外。我刚接触RabbitMQ的时候，首先安装了一下，其次从网上捞了一段发送代码发送了一条消息，比如Hello World，然后再在对侧消费到这条消息，这样我就宣告自己已经入门了。我这里也从生产-消费-以及服务端这三个方面来简单的聊一聊RabbitMQ。

首先我们先来简单的回忆一下RabbitMQ的组织结构，也可以方便的让没有接触过RabbitMQ的同学能够有个大致的印象。说到RabbitMQ，那不得不说的就是AMQP协议，AMQP协议是在RabbitMQ之前诞生的，旨在构建一个互联互通的消息中间件的开放标准。它是一个应用层协议。而RabbitMQ就是基于Erlang语言的AMQP协议的一个具体实现。

我们再来看一下这幅图，一条消息经由生产者producer发出，然后转交到交换器exchange，然后由exchange路由到队列中来做存储。消费者连接队列直接从队列中读取消息。这个我们可以套用邮政的一个例子来作说明：寄件人producer投递一个包裹到邮局exchange，由邮局负责帮忙投递到目的地queue中，之后再由收件人consumer拆件。这一套组织逻辑其实就是AMQP协议的一套逻辑，在分析RabbitMQ的时候我们完全可以套用，但是在遇到一些精细问题的时候，我们要注意在RabbitMQ中的exchange并不是一个处理模块的进程，而仅仅类似一张路由表而已，这个后面会有涉及。

下面通过一个生产者的例子来了解下RabbitMQ与AMQP之间的关系。

左边是一个生产者的简单示例代码，首先建立AMQP层面的连接Connection，然后建立信道Channel，信道Channel是依附在Connection之上的，这个拿JAVA NIO类比一下。之后基本上所有的操作都是建立在信道之上。

右边是左边代码对应的AMQP协议帧的交互，这个可以通过抓包工具，比如wireshark来查看具体的细节。首先由客户端发送Protocol Header 0-9-1的报文头与Broker建立TCP层面的连接，这个Protocol Header 0-9-1的报文头代理AMQP的版本号的信息，且这报文头是通信交互中唯一一个不是AMQP协议之内的。TCP建立起来之后，由Broker端主动发起AMQP协议层面的Connection连接请求，你也可以把这个Connection看成是一个Session。这个Connection.Start/Start-Ok主要用来搞搞协议版本以及安全之类的。之后的Connection.Tune/Tune-OK主要用来协商一些更加具体的细节，比如一个Connection中可以依附多少个信道，一帧的报文最大的长度等等。之后真正的Connection的建立完成是由客户端发起的，客户端需要挑选一个vhost（虚拟主机，多租户的一个概念）来完成Connection的建立。

之后就开启信道，然后发送消息。发送消息是由Basic.Publish来运作的，其中包含有Content-Body部分，这个Content-Body中就包含着发送的具体消息内容。最后消息发送完了就可以关闭信道和Connection。好了，我们了解了一些基本概念之后，我这里引入第一个问题：生产者如何确保消息已经正确投递到服务器Broker中呢？

AMQP协议在建立的时候就考虑到了这种情形，为此提供了事务机制的解决方案。首先通过Tx.Select/Select-Ok的交互将当前信道置为事务模式（tx mode）。然后发送一条消息，之后提交Tx.Commit，然后客户端进入阻塞等待状态等待Broker的Tx.Commit-Ok的事务确认，之后可以继续发送下一条消息。

我们在提到事务的时候，一般会有一个笼统的印象就是事务会影响性能。的确RabbitMQ也并不例外，官网原文的大致意思就说到了使用事务机制会“吸干”RabbitMQ的性能。那么有没有更好的解决方案？从AMQP层面来说，没有。但是RabbitMQ提供了另一个方案——publisher confirm即发送方确认机制。

发送方确认机制可以让生产者在发送完一条消息等待信道确认返回的时候可以继续发送下一条消息。如右图所示，首先发送方确认机制通过Confirm.Select/Select-Ok的报文交互将当前信道置为confirm模式。在发送完一条消息之后，Broker会返回一帧Basic.Ack来确认收到消息。注意事务模式和confirm模式是不能共存的，否则会报错。

听上去很完美，我们来看一下彼此的性能对比。

在所有的配置都是相同的情况下，我们看到两者之间并没有恨到的差别，发送方确认机制之比事务机制的性能提高了那么一丢丢而已，这难道是生活欺骗了我们？

这里我们再来回顾一下两者：

我们前面介绍过在事务机制中，发送完一条消息之后是阻塞等待的。在左下图中的方式，调用的waitForConfirms方法也是一种阻塞的方式。这种编程模型下，两者之间并没有什么差别。publisher confirm模式性能高一点点的原因在于它发送完一条消息的交互需要2帧报文，而事务机制需要3帧的报文。

为了提高性能，我们需要对publisher confirm模式做的就是改变编程模型，这里有两种方案：第一种，批量confirm。所谓的批量confirm是指待发送完一批消息之后再调用waitForConfirms来等待Broker端的确认，这样可以极大的减少阻塞的时间；第二步，异步confirm。所谓的异步confirm是生产者尽管发送，而消息确认是通过ConfirmListener的异步确认来实现。

我们再来比较下这4种的性能对比，包括前面的事务机制和阻塞式confirm。

可以看到批量confirm和异步confirm的方式性能提高了一大截。但是这两种模式有一个弊端就是客户端的编程模型变得复杂化了，而且需要在客户端保存状态。就拿批量confirm而言，每发送一批消息之后等待确认，在收到确认之前需要在客户端留存这一批消息，当确认失败后需要重试，进而又影响的性能。异步confirm的方式的编程模型最为复杂，至于采用哪种方式就仁者见仁智者见智了。不过为了性能和消息可靠建议采用批量或者异步confirm的方式。

那么问题由来了，这样子优化就完了嚒，图中的上面两根线中的毛刺又是为何？显然这里还有下文。

首先第一招：合并压缩，这个比较容易理解。所要考虑的就是CPU的开销，对于非CPU密集型的作业可以考虑使用下这种方式。

第二招：惰性队列，这是RabbitMQ3.6.0版本才有的共。惰性队列不管是持久化消息还是非持久化消息都是一股脑的直接存入文件系统中，这样就免去了内存的消耗。如果你本来就设置消息为持久化的，那么惰性队列简直就是“天作之合”。惰性队列不占用内存，以此也免去了消息堆积的烦恼，因为消息堆积会在一定程度上影响RabbitMQ的性能。

图中展示了普通队列和惰性队列之间的对比，很明显惰性队列的方式性能更加平稳卓越。普通队列还会受到消息堆积的烦恼，进而可能触发换页和内存阈值告警，这样就会影响性能甚至阻塞发送致使流量掉零。

不过惰性队列不是十全十美的，在消费消息的时候需要从磁盘读到内存进而再发送给消费者，这样比普通队列直接从内存中发送给消费者要稍微逊色一点，但是也差不了多少。普通队列在消息堆积的时候会换页至磁盘中，在读消息的时候也需要从磁盘读入内存再转交给消费者。

最后就是惰性队列只能在3.6.0版本以上有效，低版本无效。

第三招：HiPE, 即High Performance Erlang，高性能Erlang。很多接触RabbitMQ的同学也没听说过这个是个什么东东。这就相当于开启Java中的JIT功能，这样子可以让性能得到极大的提升。

如图所示，当前实验中，开启HiPE和不开启HiPE性能提高了近20%左右。这两次发送都触发了内存阈值告警，所以都有点流量掉零的现象，有可能数据上不是那么的完美。据官方文档显示，开启HiPE之后整体性能可以提高30%-40%左右。所付出的代价就仅仅是在启动RabbitMQ的时候多等待几十秒而已。开启这个功能也很简单，只需在配置文件中设置hipe_compile这一项为true即可。

我们再来看一下惰性队列和HiPE同时使用的效果。

可以看到惰性性队列性能提升了70%左右，而开启惰性队列和HiPE功能性能又进一步的提高了30%左右。

不过如果你使用的Erlang版本低于18.x的话，那么开启HiPE功能就相当于给自己埋了一颗定时炸弹，因为这个HiPE功能之前还并不完善，可能会致使服务性能急剧下降或者崩溃。

第四招：流控链。大多数接触过RabbitMQ的同学估计也不知道这是什么东西。

Erlang进程之间是通过消息传递来进行通信的，每个进程之间都有自己的进程邮箱mailbox来存储消息。默认情况下，Erlang并没有对进程邮箱做任何限制，所有如果有大量消息发往一个进程的时候，会导致进程邮箱过大，最终内存溢出、崩溃等。所以RabbitMQ引入了流控机制来限制进程之间的消息传递的速度。

RabbitMQ的流控机制是通过一种信用证算法来实现的，当然你可以看成是信号量。如图所示，进程B中的credit_from表示可以像下游进程C发送的消息个数，没发送一条就减一。而credit_to表示收到多少条消息之后就向上游发送一条消息，并增加相应的credit值。默认情况，RabbitMQ中设置的是{400，200}，即一个进程原本有400个credit，每向下游发送一个就减一，而下游每收到200个消息就向上游发送一条消息，并通知上游增加200个credit的值。

在RabbitMQ中，消息从生产者发出，也就是左图中的从Network流入，然后经过Connection进程，转接到Channel进程，Channel进程再查询交换器exchange得到相应的队列，进而将消息投入到队列进程中去，最后一步就是消息存储。这4个进程之间组成一个流控链的关系，对于前面3个进程Connection/Channel/Queue而言，流控的默认值为{400，200}，而最后一个rabbit_msg_store的流控阈值为{4000,800}，rabbit_msg_store进程在一个Broker中只存在一个，且被所有的队列共享，所以这个阈值会大一些。将这些默认值设置的大些可以获得性能上的一些提升，但是不能设置的过大，否则会失去流控的保护。在以前的版本中，比如rabbit_channel的流控阈值为{200,50}，github上面最新的代码显示为{400,200}。

处于流控状态的进程会标注为flow状态，从rabbitmq的管理页面上显示呈黄色状态，未处于流控状态的进程会标注为绿色的running或者灰色的idle。通过流控链可以来推断问题，当某个Connection处于流控状态，而旗下没有任何一个Channel处于流控状态，那么我们可以推断其下的某个或者多个Channel出现了性能瓶颈；当某个Connection处于流控状态，并且旗下也要有若干个channel处于流控状态，但是没有一个对应的队列处于流控状态，那么就意味着有一个或者多个队列出现了性能瓶颈；当Connection/Channel/Queue都处于流控状态，那么就以为这在消息存储时出现了性能瓶颈。

了解了这些之后我们可以执行进一步的性能优化。

经过长时间的实践应用，不难发现一般是在队列层面出现性能瓶颈。我们可以根据图中的这种方式将多个队列进程包装成一个队列进程，对于用户而言没有什么差别，但是性能却可以提升一个台阶。这里带来的问题是客户端的复杂化，需要一定程度的封装。

如上图所示，比如我们设置扩张的个数为4。首先将原本的交换器与队列的绑定转变为交换器和4个队列的绑定，绑定关系可以简单的在原有的routingKey上加一些后缀标注。

生产者在发送消息的时候可以随机或者顺序的挑选任意一个routingKey来路由发送消息。消费端在消费的时候轮询各个队列进行消费，然后暂存到本地缓存，比如简单的blockingQueue当中，然后在对外分发给消费者。优化后的效果如右下角所示，提升了很多，而且这种优化方案不会受RabbitMQ版本或者Erlang版本的限制，可谓是非常的通用。

生产者的封装比较简单，消费端的封装要变得复杂一些。消费端又分为两种消费模式：拉和推。在RabbitMQ中的拉是指单条拉取消息，一个Basic.Get就搞定；而推模式需要首先将当前信道置为投递模式，然后Broker会源源不断的发送消息给消费端，然后有客户端的回调函数去进行消费。这里就会暴露出一个问题：如果不加限制，消费端来不及消费Broker发来的消息就会造成客户端的内存持续增长直至OOM，这个问题在QueueingConsumer这个回调函数中尤为显现，在amqp-client4.0版本开始就抛弃了这个类。

不过内存溢出的问题可以通过调用Basic.Qos设置预取个数来得到有效的解决。在推模式中，Broker每发送一条消息给客户端都会计一个数，当达到设定的prefetch_count大小的时候就阻塞停止发送给对应的消费者。消费者没消费一条消息并返回Ack给消费端之后，Broker端的计数就会减一，如此就可以有效的调节消费者。那么问题就来了，这个预取个数prefetch_count设置成多少合适呢？

借用官网的一份图片（https://www.rabbitmq.com/blog/2012/05/11/some-queuing-theory-throughput-latency-and-bandwidth/）来说明下预取个数的设置。比方说Client从某个队列Queue中拿出一条消息过来需要50ms, 处理消息需要4ms，处理完之后需要返回一个ack给RabbitMQ，这样需要进一步额外的50ms以通知RabbitMQ成功确认一条消息。所以一条消息从发出到确认所需要经历104ms的往返时间。如果我们设置的预取个数prefetch_count为1，那么在这个往返行程结束之前，RabbitMQ是不会发送下一条消息给Client的。因此，Client每104ms只有4ms，或者说3.846%的时间处于忙碌状态。如果我们希望100%的时间都处于忙碌状态，那么我们可以通过往返时间除以处理时间，得到104/4=26。也就是说此种情形下，当prefetch_count的大小设置为26时，就可以让Client满负荷运行。

有人会说，把prefetch_count设置的更大点也可以让Client满载运行呀。的确如此，不过此时要考虑时延的问题。缓存的消息越多越会越发的增加消息处理的时延，对于时效性要求较高的场景来说是不可取的。从另一方面来说，如果设置的prefetch_count越大，有可能会造成对应队列Queue上的其他消费者的空闲，尤其是其他的消费者的处理能力比当前消费的处理能力强很多，比如处理一条消息只需要2ms的时间。

如果网络时延突然增加一倍怎么办？刚开始Client处理一条消息的时间只要4ms，后来由于消费端下游阻塞或者其他一些情况造成处理一条消息的时间达到40ms时又改如何应对？需要客户端做一个完美的包装，根据当前情况做一个动态调节，这样一样客户端就这一小块的东西就需要一个很繁杂的处理逻辑。对于绝大多数使用情况而言，这是得不偿失的，越多的代码意味着越多的bug隐患。回到问题的开头，绝大多数情况根据网络时延和处理速度来大致估算一下预取个数即可。

当消息大量堆积的时候，上面的考虑都将失效。首先大量堆积就以为大量的消息都不到及时处理，时延问题可以忽略；其次，也不用考虑预取个数设置的过大而造成的其他消费者空闲，因为大家都一直都有消息在流入处理。消息堆积可能会引起换页而性能下降，或者更糟糕的是触发内存告警而阻塞所有的生产者。

第一种方案是丢弃策略，即当消息超过预定的保留时间以及当前消息堆积的个数或者是内存占用而选择丢弃数据，这个可以类比于Kafka的日志保留策略，当然这种情况适合与消息可靠性要求不高、可丢弃的场景。又比如Java线程池的饱和策略中就有一种是丢弃策略。RabbitMQ本身并没有提供相应的配置和功能，这个需要外部平台的包装。

第二种方案是前面所提及的惰性队列，完全采用磁盘的空间来极大的增加堆积的能力。（一般情况下，一条服务器的磁盘容量比内存容量要大得多的多。）

第三种方案，我把它称之为“移花接木”，下面来看一张图你们就知道了。

当某个队列中的消息严重堆积时，举例：当前运行的集群cluster1中的队列queue1的消息个数超过2KW或者占用内存大小超过10GB，就可以启用shovel1将队列queue1中的消息转发至备份集群cluster2中的队列queue2中，这样可以分摊堆积的压力；当检测到队列queue1中的消息个数低于100W或者消息占用大小低于1GB时就停止shovel1，然后让原本队列queue1中的消费者慢慢处理剩余的堆积；当检测到队列queue1中的消息个数低于10W或者消息占用大小低于1GB时就开启shovel2将队列queue2中暂存的消息返还给队列queue1；当检测到队列queue1中的消息个数超过100W或者消息占用大小超过1GB时就将shovel2停掉，经过一个周期之后，再开启shovel2，超过阈值时就停掉，如此反复多次直到将队列queue2中的消息清空为止。

shovel的原理很简单，其内部有一个消费者拉取源端的数据然后暂存到本地，还有一个生产者拉取本地的消息发往目的端。

对于RabbitMQ运维层面来说，扩容和迁移是必不可少的。扩容比较简单，一般往集群中添加新的节点即可，不过RabbitMQ的扩容是垂直扩容的，新的机器节点中是没有队列进程的，只有后面新创建的队列才有可能进入这个新的节点中。迁移同样可以解决扩容的问题，将旧的集群中的数据迁移到新的容量更大的集群即可。不过RabbitMQ中的集群迁移更多的是用来解决集群故障不可短时间内修复而将所有的数据、客户端连接等迁移到新的集群中，以确保服务的可用性。

集群迁移是一个什么样的过程？

为了消息的不丢失，一般是将原有集群的producer端的连接先迁移到新的集群上，然后等待consumer消费完原有集群上的数据之后，再把consumer的连接转接到新的集群上。这里初看上去并没有什么问题，试想下如果旧集群中消息堆积比较严重，consumer的消费能力又令人堪忧，那么这个迁移等待的过程必将经历很长的时间。这里需要补充说明一下的是，消息堆积时消息中间件的一大特色功能之一，而不是什么妖魔鬼怪，消息堆积可以很好的作为大促削峰之用。不过凡事讲究适可而止，小酌怡情、大饮伤身。

Shovel可以迅速的将旧集群中的消息拉到新集群之中，进而省去了不必要的等待。

这里又展开了什么新的疑问？刚刚提到一般集群迁移是由于集群故障而做的一种应对方法，那么集群故障又有哪些呢？

集群故障有很多种类，包括机器硬件故障、脚踢电源、怒挖光缆、进程假死等，这些都是一些常见的。而RabbitMQ比较特俗的就是网络分区，也俗称为脑裂。网络分区大家应该都不陌生，比如zookeeper、keepalived等选举类的工具都有网络分区的风险，为什么说网络分区是RabbitMQ比较特殊呢？因为从RabbitMQ的设计中主动引入了网络分区，而不是被动的接受。这么说有点难以理解，我们先来说一说RabbitMQ的多副本机制——镜像队列。

如果RabbitMQ集群中只有一个Broker节点，那么该节点的失效将导致整体服务的临时性不可用，并且也可能会导致消息的丢失。可以将所有消息都设置为持久化，并且对应队列的durable属性也设置为true，但是这样仍然无法避免由于缓存导致的问题：因为消息在发送之后和被写入磁盘并执行刷盘动作之间存在一个短暂却会产生问题的时间窗。通过publisher confirm机制能够确保客户端知道哪些消息已经存入磁盘，尽管如此，一般不希望遇到因单点故障导致的服务不可用。

如果RabbitMQ集群是由多个Broker节点组成，那么从服务的整体可用性上来讲，该集群对于单点故障是有弹性的，但是同时也需要注意：尽管交换器和绑定关系能够在单点故障问题上幸免于难，但是队列和其上的存储的消息却不行，这是因为队列进程及其内容仅仅维持在单个节点之上，所以一个节点的失效表现为其对应的队列不可用。

引入镜像队列（Mirrror Queue）的机制，可以将队列镜像到集群中的其他Broker节点之上，如果集群中的一个节点失效了，队列能自动地切换到镜像中的另一个节点上以保证服务的可用性。在通常的用法中，针对每一个配置镜像的队列（以下简称镜像队列）都包含一个主节点（master）和若干个从节点（slave）

RabbitMQ集群中的所有节点都会备份所有的元数据信息，包括队列元数据（队列的名称和属性）、交换器、绑定关系、vhost等。但是队列进程本身是不会备份的，基于存储空间和性能的考虑，在RabbitMQ集群中创建队列，集群只会在单个节点而不是在所有节点上创建队列的进程，这个队列的进程就是前面在流控中所讲述的rabbit_amqpqueue_process。如此只有队列的宿主节点知道队列的所有信息，所有其他非宿主节点只知道队列的元数据和指向该队列存在的那个节点的指针。因此当集群节点崩溃时，该节点的队列进程和关联的绑定都会消失，与之对应的消息和消费者也就没了。

为了防止队列以及队列中的消息因Broker单点故障而丢失，RabbitMQ引入镜像队列的机制，这样可以将队列以镜像副本的形式分布于其他的Broker之上。如果集群中的一个Broker失效了，队列能自动地切换到镜像中的另一个节点上以保证服务的可用性。选举是选择镜像中资历最老的那个，也就是同步消息最全的那个为新的master。

在镜像队列的模式中，消息会被发往master和slave上，如果此时master挂掉，消息还会在slave上，这样slave提升为master的时候消息也不会丢失。除了发送消息之外都是由master先处理，之后再由master将处理的结果广播给各个slave，而这个广播的过程是一个环形的广播，比如收到消息之后需要返回给客户端Ack（这个在开头生产者确认机制publisher confirm中有提到过），这个不是由master直接返回给客户端的，而是又master先将Ack信号传递给slave1,slave1Ack之后再传递给slave2，slave2 Ack之后再传回给master，这样在镜像队列中转完一圈之后即所有副本都Ack之后，才由master返回给客户端。

那么如果客户端连接的不是master而是其中的一个slave呢？就以consumer为例，如果消费者与slave建立连接并进行订阅消费，其实质上都是从master上获取消息，只不过看似是从slave上消费而已。比如消费者与slave建立了TCP连接之后执行一个Basic.Get的操作，就是拉模式拉一条数据，那么首先是slave将Basic.Get请求发往master，然后再由master准备好数据返回给slave，最后由slave投递给消费者。

疑问：大多数的读写压力都落到了master之上，那么这样负载是否会做不到有效的均衡呢？

所以在创建队列的时候一般是刻意的将master节点均摊到各个Broker之上。如果你了解过Kafka，RabbitMQ的队列和Kafka的分区partition有相似之处，Kafka的多副本机制中partition分为leader副本和follower副本，而所有的读写也都只在leader副本中完成。

默认情况下，客户端与哪个Broker相连，队列就创建到哪个Broker之上。我们这里可以通过对客户端的一些包装来达到分摊队列的目的。比如通过随机法随机选取集群中的任意一个Broker建立连接然后再创建队列。还有比如轮询、哈希地址、最小连接数等待方法；亦或者是通过第三方工具如HAProxy或者LVS来实现负载均衡的目的（并不建议）。不过RabbitMQ本身也考虑到这方面的顾虑，在较新的版本中可以通过在rabbitmq.config配置文件中配置queue-master-locator的参数，或者在客户端连接时设置x-queue-master-locator的参数就可以方便的来实现负载均衡，不过这个配置默认的是client-local即创建队列的时候是在所连接的Broker的本地创建的，可以设置为min-masters，这样可以灵活的在具有最少master镜像的Broker上创建队列；或者选择random也行，这样可以随机选择一个Broker创建队列。

集群中的某个Broker异常崩溃后，其上所有的master都会失效，再新的一轮选举之后都被分摊到其他的Broker之上，当崩溃的Broker恢复之后，其上一个master都不会存在，这样负载就会失衡，那么请问有什么办法可以解决？（这里问题这里不做详细陈述，可以私信留言探讨。）

(具体讲解略过~图中是网络分区的思维导图，所有心法都列于其上，各位施主请自取~~)

更多相关文章

随机推荐