消息队列总结（4）- RabbitMQ Kafka RocketMQ高性能方案

1.RabbitMQ的高性能解决方案

1.1 发布确认机制

RabbitMQ提供了3种生产者发布确认的模式：

简单模式（Simple Mode）：生产者发送消息后，等待服务器确认消息已经被接收。这种模式下，生产者发送消息后会阻塞，直到收到服务器确认消息。如果服务器在确认消息前崩溃，生产者会重新发送消息。同步等待确认，实现简单，但是吞吐量十分有限。
批量模式（Batch Mode）：生产者发送一批消息后，等待服务器一次性确认这批消息已经被接收。这种模式相比简单模式有更高的吞吐量，因为确认是批量进行的。批量同步等待确认，实现简单，吞吐量较大，但是很难找出未确认的消息，其中一个失败后需要把一个批次都重试。
异步模式（Asynchronous Mode）：生产者发送消息后，不会等待服务器的确认消息。而是通过回调函数来处理确认和错误信息。这种模式适用于对消息可靠性要求不高的场景，可以提高生产者的性能。可靠性和性能最好，在出现未确认消息时容易处理，但是实现困难。

RabbitMQ的默认分发方式是轮询分发，轮询分发的问题是会导致消费快的消费者空闲，消费慢的消费者一直干活。为了解决这个问题，RabbitMQ引入了不公平分发机制，可以把任务分发给空闲的消费者。

Channel channel = connection.createChannel();
channel.basicQos(1)

上面案例中方法basicQos的参数PrefetchCount（案例中等于1）是最大传输信息数，当消息由消费者消费完成之后，再次从Queue中获取消息，达到预取值。

通过预取值的机制可以减少消费者与磁盘之间的交换次数，从而提升消费者的处理能力。

Producer会为每个Partition创建一个双端队列来缓存客户端的消息，队列中的每个元素是PorducerBatch，PorducerBatch的每个元素就是客户端要发送的Msg。

接下来sender线程工作机制是：

寻找ReadyNode：sender到消息累加器中轮询存在哪些Node已经准备好的ProducerBatch，只要一个Node有任何一个ProducerBatch准备好，这个Node就会被认为是ReadyNode。
创建Request：拿到所有的ReadyNode，寻找其中准备好的ProducerBatch，对于一个Node下的ProducerBatch打包成一个Request，其中一个Request最多包含的ProducerBatch由max.request.size控制。
发起通讯：然后每个Request通过Selector发起通讯。

sender把消息发送到Broker有两个条件：

磁盘通查查询一条数据的过程如下：

从上面的过程可以看出，如果我们查询/写入一条数据是随机在磁盘的一个位置，那么整个过程会比较耗时。对于Kafka来说，采用的策略是使用顺序IO，这样就可以避免寻址的过程，直接操作对数据的读/写操作。

传统情况下，从磁盘读取数据，并通过网络发出去需要2次CPU copy和2次DMA copy：

数据读取过程：DMA执行了一次数据拷贝，数据从磁盘拷贝到内核空间。cpu再将数据从内核空间拷贝到用户空间（用户缓冲区）。
数据发送过程：cpu发生第三次数据拷贝，由cpu将数据从用户空间拷贝至内核空间(socket缓冲区)，DMA执行第四次数据拷贝，将数据从内核空间写到网卡。

Linux2.4+的Linux系统支持了sendfile + DMA Gather：

这样只有两次上下文切换和两次DMA copy极大的减少了系统开支。

RocketMQ在高性能上与Kafka类似，使用异步、批量、零拷贝的机制，来实现高吞吐量。具体RocketMQ的异步机制如下：

数据写入CommitLog：Broker接收来自Producer发出的消息，获取CommitLog最新offset，并往CommitLog对应ByteBuffer追加数据。
异步写磁盘：Broker通过同步/异步的方式写入到磁盘。若为异步写入磁盘，则是把数据写入OS的Page Cache就给Producer返回ACK，后台线程异步把Page Cache的数据写的磁盘。
异步复制：Broker通过同步/异步的方式进行Master/Slave之间的数据同步。若为异步复制，则是数据写入Master成功即视为成功，再后台异步同步至其他Slave。
异步写ConsumerQueue：后台线程轮询CommitLog的offset是否发生变化，若发生变化，则计算CommitLog对应消息的commitLog Offset、size、Message Tag HashCode写入ConsumerQueue。
异步写IndexFIle：写入ConsumerQueue后，再将消息Key Hash、commitLog Offset、TimeStamp、Next Index Offset写入到到IndexFile。

（在RocketMQ中使用的批量发送、零拷贝等机制在上面已讲过，不再重复陈述）