消息队列RocketMQ\kafka的测试对比与RocketMQ使用扩展

消息队列的测试对比与RocketMQ使用扩展

1.消息队列的选择：核心业务弃用kafka

kafka用的是发布/订阅模式，一个topic的订阅方会有很多，涉及到的下游业务也就非常多，没办法一口气直接替换kafka，迁移到新的一个消息队列服务上。

我们用的是kafka 0.8.2版本，业务在使用kafka的时候，集群数据写入抖动非常严重，会有数据写失败。

主要原因有两点：

1. 随着业务增长，topic的数据增多，集群负载增大，性能下降。

2. kafka有个bug，会导致副本重新复制，复制的时候有大量的读，机械盘的存储方式，将导致磁盘IO过大，影响写入。

所以我们首先需要解决上面的业务方消息生产失败的问题

如上图，我们的方案是加了一层代理，然后利用codis作为缓存，当kafka出现不可写入的时候，先把数据写入到codis中，然后延时进行重试，直到写成功为止。这样解决kafka不定期写入失败的问题。

2.消息队列的选择：核心业务选择RocketMQ

2.1 测试-不同topic数量下的吞吐测试

测试环境及版本：

Kafka 0.8.2

RocketMQ 3.4.6

1.0 Gbps Network

16 threads

上图是Kafka和RocketMQ在不同topic数量下的吞吐测试，on 表示消息发送的时候，同时进行消息消费，off表示仅进行消息发送。横坐标代表每秒消息数，纵坐标是测试case。同时覆盖了有无消费，和不同消息体的场景。一共8组测试数据，每组数据分别在topic个数为16、32、64、128、256时获得的，每个topic包括8个partition。下面四组数据是发送消息大小为128字节的情况，上面四种是发送2k消息大小的情况。

第一组数据，kafka开启消费，每条消息大小为2048字节。看图可知，当topic数量增加到256 时，吞吐极具下降。

第二组数据是RocketMQ。当topic增大之后，影响非常小。

第三组和第四组，是上面两组关闭了消费的情况。可以看到整体吞吐量会高那么一点点。

下面的四组跟上面的区别是使用了128字节的小消息体。

可以看到，kafka吞吐受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超过1w tps 就超过1ms了。kafka的延迟明显更高。

2.3消息队列的选择：核心业务选择RocketMQ

经过我们的调研和测试，我们决定采用RocketMQ。

确定好基本框架之后，我们后面就主要围绕三个方向做。

一个是迁移。把公司之前所有的队列环境进行迁移。

第二个就是功能迭代和性能上的优化。

最后一个就是服务化，业务直接通过平台界面来申请资源，申请到之后直接使用。

3. 如何构建自己的消息队列服务

3.1 问题与挑战

面临的挑战（顺时针看）：

• 需要支持多种客户端语言。需要支持PHP、GO、Java、C++。

• 技术力量储备不足，开发人员少。

• 技术知识储备不足，决定用RocketMQ，但是没进行更深入研究，没看过源码。

• 上线周期紧，线上的消息队列还有问题。

• 保证系统高并发环境下高可用。

使用RocketMQ时的两个问题：

• 客户端语言主要支持Java。而我们还需要支持PHP、Go、C++，RocketMQ目前提供的Go的sdk我们测的有一些问题。

• 很难保证系统高并发环境下的稳定性。RocketMQ功能特别多，如tag、property、消费过滤、RETRY topic、死信队列、延迟消费之类的功能，非常丰富。

解决问题的主要策略就是坚持KISS原则（Keep it simple, stupid），保持简单，先解决最主要的问题，让消息能够流转起来。

解决办法：

• 使用Thrift RPC框架来解决跨语言的问题。

• 简化调用接口。可以认为只有两个接口，send用来生产，pull用来消费。

然后我们把其他主要逻辑都放在了proxy这一层来做，比如权限认证、消息过滤、限流、格式转化之类的。这样，我们就不需要把很多功能用各种语言都写一遍，能尽可能地简化客户端的实现逻辑。

3.2 迁移方案

确定架构之后，然后就要确定迁移方案。

因为下游的数据消费方很多，上游的数据没法做到一刀切流量，这就会导致整个迁移的周期特别长。所以迁移这个事情，在pub-sub的消息模型下，会比较复杂。

为了加快迁移的效率，尽可能地减少业务迁移的负担，我们在proxy层提供了双写和双读的功能。

• 双写：Procucer Proxy同时写RocketMQ和kafka。

• 双读：Consumer Proxy同时从RocketMQ和kafka消费数据。

基于这两个功能，我们就制定了两种迁移方案。

3.2.1 双写

生产端双写，同时往kafka和rocketmq写同样的数据，保证两边在整个迁移过程中都有同样的全量数据。有相同的数据，下游的业务也就可以迁移。

如果消费端不关心丢数据，可以直接切换，切完直接更新消费进度。

如果需要保证消费必达，可以先在Consumer Proxy设置消费进度，消费客户端保证没有数据堆积后再去迁移，这样会有一些重复消息。

生产端的双写也有两种方案：

1. 客户端双写，如下图：

业务那边不停原来的kafka 客户端。只是加上我们的客户端，往RocketMQ里追加写。

这种方案在整个迁移完成之后，业务还需要把老的写入停掉。相当于两次上线。

1. 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，这样就能实现没有大量重复数据的效果。

所以这个方案对重复消费比较敏感的业务会比较适合的。

这个方案的整个过程中，消费方和生产方都只需要改造一遍客户端，上一次线就可以完成。

4. RocketMQ扩展改造

首先是主从的自动切换。

目前开源版本的RocketMQ broker 是没有主从自动切换的。如果master挂了，就写不进去了。然后slave只能提供只读的功能。

所以，我们就自己加了一套主从自动切换的功能。

当然如果在多个主节点上都创建topic，虽然不会完全写不进去，但对单分片顺序消费的场景会产生影响。

第二个是批量生产的功能。RocketMQ 4.0之后的版本是支持批量生产功能的。但是限制了，只能是同一个ConsumerQueue的。所以我们就扩展了一下，让proxy能够支持不同topic、不同consume queue。

原理上只是在传输的时候，把topic和consumerqueue的信息都编码进去。

第三个，目前RocketMQ 单机能够支持的topic数量，基本是在几万的量级，如果继续增加，元信息的管理就会非常耗时，系统的整个吞吐就会非常大。所以我们就改造了一下元信息管理部分，让RocketMQ单机能够支撑的topic数量达到了百万。

我们还扩展了一些功能，比如集成了我们公司内部的一些监控和部署工具。

5. RocketMQ使用经验

5.1 读老数据的问题

我们都知道，RocketMQ的数据是要落盘的，一般只有最新写入的数据才会在PageCache中。比如下游消费数据，因为一些原因停了一天之后，又突然起来消费数据。这个时候就需要读磁盘上的数据。

然后RocketMQ的消息体是全部存储在一个append only的 commitlog 中的。如果这个集群中混杂了很多不同topic的数据的话，要读的两条消息就很有可能间隔很远。最坏情况就是一次磁盘IO读一条消息。这就基本等价于随机读取了。如果磁盘的IOPS（Input/Output Operations Per Second）扛不住，还会影响数据的写入，这个问题就严重了。

值得庆幸的是，RocketMQ提供了自动从Slave读取老数据的功能。这个功能主要由slaveReadEnable这个参数控制。默认是关的（slaveReadEnable = false by default）。推荐把它打开，主从都要开。

这个参数打开之后，在客户端消费数据时，会判断，当前读取消息的物理偏移量跟最新的位置的差值，是不是超过了内存容量的一个百分比（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发出请求。