江海挺：
滴滴出行行列步队卖力人，Apache RocketMQ Contributor，大学毕业后一贯在做行列步队领域干系的技能、产品和做事，积累了丰富的实践履历，沉淀了不少关于行列步队的思考。
滴滴出行的技能选型
1.1 历史
初期，公司内部没有专门的团队掩护行列步队做事，以是行列步队利用办法较多，紧张以Kafka为主，有业务直连的，也有通过独立的做事转发的。其余有一些团队也会用RocketMQ、Redis的list，乃至会用比较非主流的beanstalkkd。导致的结果便是，比较混乱，无法掩护，资源利用也很摧残浪费蹂躏。
1.2 为什么弃用 Kafka
一个核心业务在利用Kafka的时候，涌现了集群数据写入抖动非常严重的情形，常常会有数据写失落败。
紧张有两点缘故原由：
随着业务增长，Topic的数据增多，集群负载增大，性能低落；我们用的是Kafka0.8.2那个版本，有个bug，会导致副本重新复制，复制的时候有大量的读，我们存储盘用的是机器盘，导致磁盘IO过大，影响写入。以是我们决定做自己的行列步队做事。
首先须要办理业务方生产失落败的问题。由于这个Kafka用的是发布/订阅模式，一个topic的订阅方会有很多，涉及到的下贱业务也就非常多，没办法一口气直接更换Kafka，迁移到新的一个行列步队做事上。以是我们当时的方案是加了一层代理，然后利用codis作为缓存，办理了Kafka不定期写入失落败的问题，如上图。当后面的Kafka涌现不可写入的时候，我们就会先把数据写入到codis中，然后延时进行重试，直到写成功为止。
1.3 为什么选择 RocketMQ
经由一系列的调研和测试之后，我们决定采取RocketMQ，详细缘故原由在后面会先容。
为了支持多措辞环境、办理一些迁移和某些业务的分外需求，我们又在消费侧加上了一个代理做事。然后形成了这么一个核心框架。业务端只跟代理层交互。中间的引擎，卖力的核心存储。在之前的基本框架之后，我们后面就紧张环绕三个方向做。
迁移，把之条件到的所有五花八门的行列步队环境，全部迁移到我们上面。这里面的迁移方案后面会跟大家先容一下。功能迭代和整天机能上的优化。做事化，业务直接通过平台界面来申请资源，申请到之后直策应用。1.4 演进中的架构
这张图是我们行列步队做事的一个比较新的现状。先纵向看，上面是生产的客户端，包括了7种措辞。然后是我们的生产代理做事。在中间的是我们的存储层。目前紧张的存储引擎是RocketMQ。然后还有一些在迁移过程中的Kafka。另一个是Chronos，它是我们延迟的一个存储引擎。
再下面便是消费代理。消费代理同样供应了多种措辞的客户端，还支持多种协议的主动推送功能，包括HTTP 协议 RESTful办法。结合我们的groovy脚本功能，还能实现将直接转存到Redis、Hbase和HDFS上。此外，我们还在陆续接入更多的下贱存储。
除了存储系统之外，我们也对接了实时打算平台，例如Flink，Spark，Storm，左边是我们的用户掌握台和运维掌握台。这个是我们做事化的重点。用户在须要利用行列步队的时候，就通过界面申请Topic，填写各种信息，包括身份信息，的峰值流量，大小，格式等等。然后消费方通过我们的界面，就可以申请消费。
运维掌握台，紧张卖力我们集群的管理，自动化支配，流量调度，状态显示之类的功能。末了所有运维和用户操作会影响线上的配置，都会通过ZooKeeper进行同步。
为什么选择RocketMQ
我们环绕以下两个纬度进行了比拟测试，结果显示RocketMQ的效果更好。
2.1 测试-topic数量的支持
测试环境：Kafka 0.8.2，RocketMQ 3.4.6，1.0 Gbps Network，16 threads
测试结果如下：这张图是Kafka和RocketMQ在不同topic数量下的吞吐测试。横坐标是每秒数，纵坐标是测试case。同时覆盖了有无消费，和不同体的场景。一共8组测试数据，每组数据分别在Topic个数为16、32、64、128、256时得到的，每个topic包括8个Partition。下面四组数据是发送大小为128字节的情形，上面四种是发送2k大小的情形。on 表示发送的时候，同时进行消费，off表示仅进行发送。
先看最上面一组数据，用的是Kafka，开启消费，每条大小为2048字节可以看到，随着Topic数量增加，到256 Topic之后，吞吐极具低落。第二组是是RocketMQ。可以看到，Topic增大之后，影响非常小。第三组和第四组，是上面两组关闭了消费的情形。结论基本类似，整体吞吐量会高那么一点点。
下面的四组跟上面的差异是利用了128字节的小体。可以看到，Kafka吞吐受Topic数量的影响特殊明显。比拟来看，虽然topic比较小的时候，RocketMQ吞吐较小，但是基本非常稳定，对付我们这种共享集群来说比较友好。
2.2 测试-延迟
Kafka测试环境：Kafka 0.8.2.2，topic=1/8/32，Ack=1/all，replica=3
测试结果：
上面的一组的3条线对应Ack=3，须要3个备份都确认后才完成数据的写入。下面的一组的3条线对应Ack=1，有1个备份收到数据后就可以完成写入。可以看到下面一组只须要主备份确认的写入，延迟明显较低。每组的三条线之间紧张是Topic数量的差异，Topic数量增加，延迟也增大了。
RocketMQ测试环境：
RocketMQ 3.4.6，brokerRole=ASYNC/SYNC_MASTER, 2 Slave，
flushDiskType=SYNC_FLUSH/ASYNC_FLUSH
测试结果：
上面两条是同步刷盘的情形，延迟相比拟较高。下面的是异步刷盘。橙色的线是同步主从，蓝色的线是异步主从。然后可以看到在副本同步复制的情形下，即橙色的线，4w的TPS之内都不超过1ms。用这条橙色的线和上面Kafka的图中的上面三条线横向比较来看，Kafka超过1w TPS 就超过1ms了。Kafka的延迟明显更高。
如何构建自己的行列步队
3.1 问题与寻衅
面临的寻衅（顺时针看）
客户端措辞，须要支持PHP、Go、Java、C++；只有3个开拓职员；决定用RocketMQ，但是没看过源码；上线韶光紧，线上的Kafka还有问题；可用性哀求高。利用RocketMQ时的两个问题：
客户端措辞支持不全，以Java为主，而我们还须要支持PHP、Go、C++；功能特殊多，如tag、property、消费过滤、RETRYtopic、去世信行列步队、延迟消费之类的功能，但这对我们稳定性掩护来说，寻衅非常大。针对以上两个问题的办理办法，如下图所示：
利用ThriftRPC框架来办理跨措辞的问题；简化调用接口。可以认为只有两个接口，send用光降盆，pull用来消费。
紧张策略便是坚持KISS原则（Keep it simple, stupid），保持大略，先办理最紧张的问题，让能够流转起来。然后我们把其他紧张逻辑都放在了proxy这一层来做，比如限流、权限认证、过滤、格式转化之类的。这样，我们就能尽可能地简化客户真个实现逻辑，不须要把很多功能用各种措辞都写一遍。
3.2 迁移方案
架构确定后，接下来是我们的一个迁移过程。
迁移这个事情，在pub-sub的模型下，会比较繁芜。由于下贱的数据消费方可能很多，上游的数据没法做到一刀切流量，这就会导致全体迁移的周期特殊长。然后我们为了尽可能地减少业务迁移的包袱，加快迁移的效率，我们在Proxy层供应了双写和双读的功能。
双写：ProcucerProxy同时写RocketMQ和Kafka；双读：ConsumerProxy同时从RocketMQ和Kafka消费数据。有了这两个功能之后，我们就能供应以下两种迁移方案了。
3.2.1 双写
生产端双写，同时往Kafka和RocketMQ写同样的数据，担保两边在全体迁移过程中都有同样的全量数据。Kafka和RocketMQ有相同的数据，这样下贱的业务也就可以开始迁移。如果消费端不关心丢数据，那么可以直接切换，切完直接更新消费进度。如果须要担保消费必达，可以先在ConsumerProxy设置消费进度，消费客户端担保没有数据堆积后再去迁移，这样会有一些重复，一样平常客户端会担保消费处理的幂等。
生产真个双写实在也有两种方案：
客户端双写，如下图：
业务那边一直原来的kafka 客户端。只是加上我们的客户端，往RocketMQ里追加写。这种方案在全体迁移完成之后，业务还须要把老的写入停掉。相称于两次上线。
Producer Proxy双写，如下图：
业务方直接切换生产的客户端，只往我们的proxy上写数据。然后我们的proxy卖力把数据复制，同时写到两个存储引擎中。这样在迁移完成之后，我们只须要在Proxy上关掉双写功能就可以了。对生产的业务方来说是无感知的，生产方全程只须要改造一次，上一下线就可以了。
以是表面看起来，该当还是第二种方案更加大略。但是，从整体可靠性的角度来看，一样平常还是认为第一种相对高一点。由于客户端到Kafka这一条链路，业务之前都已经跑稳定了。一样平常不会出问题。但是写我们Proxy就不一定了，在接入过程中，是有可能涌现一些利用上的问题，导致数据写入失落败，这就对业务方测试质量的哀求会高一点。然后消费的迁移过程，实在风险是相比拟较低的。出问题的时候，可以立即回滚。由于它在老的Kafka上消费进度，是一贯保留的，而且在迁移过程中，可以认为是全量双消费。
以上便是数据双写的迁移方案，这种方案的特点便是两个存储引擎都有相同的全量数据。
3.2.2 双读
特点：担保不会重复消费。对付P2P 或者消费下贱不太多，或者对重复消费数据比较敏感的场景比较适用。
这个方案的过程是这样的，消费先切换。全部迁移到到我们的Proxy上消费，Proxy从Kafka上获取。这个时候RocketMQ上没有流量。但是我们的消费Proxy担保了双消费，一旦RocketMQ有流量了，客户端同样也能收到。然后生产方改造客户端，直接切流到RocketMQ中，这样就完成了全体流量迁移过程。运行一段韶光，比如Kafka里的数据都过期之后，就可以把消费Proxy上的双消费关了，下掉Kafka集群。
全体过程中，生产直接切流，以是数据不会重复存储。然后在消费迁移的过程中，我们消费Proxy上的group和业务原有的group可以用一个名字，这样就能实现迁移过程中自动rebalance，这样就能实现没有大量重复数据的效果。以是这个方案对重复消费比较敏感的业务会比较适宜的。这个方案的全体过程中，消费方和生产方都只须要改造一遍客户端，上一次线就可以完成。
RocketMQ扩展改造
说完迁移方案，这里再大略先容一下，我们在自己的RocketMQ分支上做的一些比较主要的事情。
首先一个非常主要的一点是主从的自动切换。
熟习RocketMQ的同学该当知道，目前开源版本的RocketMQ broker 是没有主从自动切换的。如果你的Master挂了，那你就写不进去了。然后slave只能供应只读的功能。当然如果你的topic在多个主节点上都创建了，虽然不会完备写不进去，但是对单分片顺序消费的场景，还是会产生影响。以是呢，我们就自己加了一套主从自动切换的功能。
第二个是批量生产的功能。
RocketMQ4.0之后的版本是支持批量生产功能的。但是限定了，只能是同一个ConsumerQueue的。这个对付我们的Proxy做事来说，不太友好，由于我们的proxy是有多个不同的topic的，以是我们就扩展了一下，让它能够支持不同Topic、不同Consume Queue。事理上实在差不多，只是在传输的时候，把Topic和Consumer Queue的信息都编码进去。
第三个，元信息管理的改造。
目前RocketMQ单机能够支持的Topic数量，基本在几万这么一个量级，在增加上去之后，元信息的管理就会非常耗时，对全体吞吐的性能影响相对来说就会非常大。然后我们有个场景又须要支持单机百万旁边的Topic数量，以是我们就改造了一下元信息管理部分，让RocketMQ单机能够支撑的Topic数量达到了百万。
后面一些就不太主要了，比如集成了我们公司内部的一些监控和支配工具，修了几个bug，也给提了PR。最新版都已经fix掉了。
RocketMQ利用履历
接下来，再大略先容一下，我们在RocketMQ在利用和运维上的一些履历。紧张是涉及在磁盘IO性能不足的时候，一些参数的调度。
5.1 读老数据的问题
我们都知道，RocketMQ的数据是要落盘的，一样平常只有最新写入的数据才会在PageCache中。比如下贱消费数据，由于一些缘故原由停了一天之后，又溘然起来消费数据。这个时候就须要读磁盘上的数据。然后RocketMQ的体是全部存储在一个append only的 commitlog 中的。如果这个集群中殽杂了很多不同topic的数据的话，要读的两条就很有可能间隔很远。最坏情形便是一次磁盘IO读一条。这就基本等价于随机读取了。如果磁盘的IOPS（Input/Output Operations Per Second）扛不住，还会影响数据的写入，这个问题就严重了。
值得光彩的是，RocketMQ供应了自动从Slave读取老数据的功能。这个功能紧张由slaveReadEnable这个参数掌握。默认是关的（slaveReadEnable = false bydefault）。推举把它打开，主从都要开。这个参数打开之后，在客户端消费数据时，会判断，当前读取消息的物理偏移量跟最新的位置的差值，是不是超过了内存容量的一个百分比（accessMessageInMemoryMaxRatio= 40 by default）。如果超过了，就会见告客户端去备机上消费数据。如果采取异步主从，也便是brokerRole即是ASYNC_AMSTER的时候，你的备机IO打爆，实在影响不太大。但是如果你采取同步主从，那还是有影响。以是这个时候，最好挂两个备机。由于RocketMQ的主从同步复制，只要一个备机相应了确认写入就可以了，一台IO打爆，问题不大。
5.2 过期数据删除
RocketMQ默认数据保留72个小时（fileReservedTime=72）。然后它默认在凌晨4点开始删过期数据（deleteWhen=\公众04\公众）。你可以设置多个值用分号隔开。由于数据都是定时删除的，以是在磁盘充足的情形，数据的最长保留会比你设置的还多一天。又由于默认都是同一韶光，删除一整天的数据，如果用了机器硬盘，一样平常磁盘容量会比较大，须要删除的数据会特殊多，这个就会导致在删除数据的时候，磁盘IO被打满。这个时候又要影响写入了。
为理解决这个问题，可以考试测验多个方法，一个是设置文件删除的间隔，有两个参数可以设置，
deleteCommitLogFilesInterval = 100（毫秒）。每删除10个commitLog文件的韶光间隔；deleteConsumeQueueFilesInterval=100（毫秒）。每删除一个ConsumeQueue文件的韶光间隔。其余一个便是增加删除频率，把00-23都写到deleteWhen，就可以实现每个小时都删数据。
5.3 索引
默认情形下，所有的broker都会建立索引（messageIndexEnable=true）。这个索引功能可以支持按照的uniqId，的key来查询体。索引文件实现的时候，实质上也便是基于磁盘的个一个hashmap。如果broker上数量比较多，查询的频率比较高，这也会造成一定的IO负载。以是我们的推举方案是在Master上关掉了index功能，只在slave上打开。然后所有的index查询全部在slave上进行。当然这个须要大略修正一下MQAdminImpl里的实现。由于默认情形下，它会向Master发出要求。
作者：中间件小哥
