关于消息中间件:快手基于RocketMQ的在线消息系统建设实践

2次阅读

共计 4935 个字符，预计需要花费 13 分钟才能阅读完成。

简介： 快手须要建设一个次要面向在线业务的音讯零碎作为 Kafka 的补充，低提早、高并发、高可用、高牢靠的分布式消息中间件 RocketMQ 正是咱们所需的。

作者：黄理

黄理，10 多年软件开发和架构教训，热衷于代码和性能优化，开发和参加过多个开源我的项目。曾在淘宝任业务架构师多年，以后在快手负责在线音讯零碎建设工作。

在引入 RocketMQ 之前，快手曾经在大量的应用 Kafka 了，但并非所有状况下 Kafka 都是最合适的，比方以下场景：

业务心愿个别生产失败当前能够重试，并且不梗塞后续其它音讯的生产。
业务心愿音讯能够提早一段时间再投递。
业务须要发送的时候保障数据库操作和音讯发送是统一的（也就是事务发送）。
为了排查问题，有的时候业务须要肯定的单个音讯查问能力。

为了应答以上这类场景，咱们须要建设一个次要面向在线业务的音讯零碎，作为 Kafka 的补充。在考查的一些消息中间件中，RocketMQ 和业务需要匹配度比拟高，同时部署构造简略，应用的公司也比拟多，于是最初咱们就采纳了 RocketMQ。

在一个已有的体系内落地一个开软软件，通常大略有两种形式：

形式一，在开源软件的根底上做深度批改，很容易实现公司内须要的定制性能。但和社区开源版本各奔前程，当前如何降级？

形式二，尽量不批改社区版本（或缩小不兼容的批改），而是在它的外围或者下层进一步包装来实现公司外部须要的定制性能。

注：上图形式一的图画的比拟极其，实际上很多公司是形式一、形式二联合的。

咱们抉择了形式二。最早的时候，咱们应用的是 4.5.2 版本，起初社区 4.7 版本大幅减小了同步复制的提早，正好咱们的部署模式就是同步复制，于是就很轻松的降级了 4.7 系列，享受了新版本的红利。

在部署集群的时候，还会面临很多部署策略的抉择：

• 大集群 vs 小集群

• 抉择正本数

• 同步刷盘 vs 异步刷盘

• 同步复制 vs 异步复制

• SSD vs 机械硬盘

大集群会有更好的性能弹性，而小集群具备更好的隔离型，此外小集群能够不须要跨可用区 /IDC 部署，所以会有更好的健壮性，咱们十分看重稳定性，因而抉择了小集群。集群同步复制异步刷盘，首选 SSD。

如上所述，咱们没有在 Rocketmq 外面做深度批改，所以须要提供一个 SDK 来提供公司内的须要的定制性能，这个 SDK 大略是这样的：

对外只提供最根本的 API，所有拜访必须通过咱们提供的接口。简洁的 API 就像冰山的一个角，除了对外的简略接口，上面所有的货色都能够降级更换，而不会毁坏兼容性。

业务开发起来也很简略，只有须要提供 Topic（全局惟一）和 Group 就能够生产和生产，不必提供环境、Name Server 地址等。SDK 外部会依据 Topic 解析出集群 Name Server 的地址，而后连贯相应的集群。生产环境和测试环境环境会解析出不同的地址，从而实现了隔离。

上图分为 3 层，第二层是通用的，第三层才对应具体的 MQ 实现，因而，实践上能够更换为其它消息中间件，而客户端程序不须要批改。

SDK 外部集成了热变更机制，能够在不重启 client 的状况下做动静配置，比方下发路由策略（更换集群 name server 的地址，或者连贯到别的集群去），Client 的线程数、超时工夫等。通过 maven 强制更新机制，能够保障业务应用的 SDK 基本上是最新的。

所有的 Topic 默认都调配到两个可用区，生产者和消费者会同时连贯至多两个独立集群（散布在不同的可用区），如下图：

生产者同时连贯两个集群，如果可用区 A 呈现故障，流量就会主动切换到可用区 B 的集群 2 去。咱们开发了一个小组件来实现自适应的集群负载平衡，它蕴含以下能力：

• 千万级 OPS

• 灵便的权重调整策略

• 健康检查反对 / 事件告诉

• 并发度管制（主动升高响应慢的服务器的申请数）

• 资源优先级（相似 Envoy，实现本地机房优先，或是被调服务器很多的时候选取一个子集来调用）

• 主动优先级治理

• 增量热变更

实际上它并不仅仅用于音讯生产者，而是一个通用的主调方负载平衡类库，能够在 github 上找到：

https://github.com/PhantomThief/simple-failover-java

外围的 SimpleFailover 接口和 PriorityFailover 类没有传递第三方依赖，非常容易整合。

提早音讯是十分重要的业务性能，不过 RocketMQ 内置的提早音讯只能反对几个固定的提早级别，所以咱们又开发了独自的 Delay Server 来调度提早音讯：

上图这个构造没有间接将提早音讯发到 Delay Server，而是更换 Topic 当前存入 RocketMQ。这样的益处是能够复用现有的音讯发送接口（以及下面的所有扩大能力）。对业务来说，只须要在结构音讯的时候额定指定一个延迟时间字段即可，其它用法都不变。

RocketMQ 4.3 版本当前反对了事务音讯，能够保障本地事务和生产发送同时胜利或者失败，对于一些业务场景很有帮忙。事务音讯的用法和原理有很多材料，这里就不细述了。但对于事务音讯的实际网上材料较少，咱们能够给出一些倡议。

首先，事务音讯性能始终在不断完善，应该应用最新的版本，至多是 4.6.1 当前的版本，能够防止很多问题。

其次，事务音讯性能是不如一般音讯的，它在外部实际上会生成 3 个音讯（一阶段 1 个，二阶段 2 个），所以性能大概只有一般音讯的 1 /3，如果事务音讯量大的话，要做好容量布局。回查调度线程也只有 1 个，不要用极限压力去考验它。

最初有一些参数注意事项。在 broker 的配置中：

transientStorePoolEnable 这个参数必须放弃默认值 false，否则会有重大的问题。
endTransactionThreadPoolNums 是事务音讯二阶段解决线程大小，sendMessageThreadPoolNums 则指定一阶段解决线程池大小。如果二阶段的处理速度跟不上一阶段，就会造成二阶段音讯失落导致大量回查，所以倡议 endTransactionThreadPoolNums 应该大于 sendMessageThreadPoolNums，倡议至多 4 倍。
useReentrantLockWhenPutMessage 设置为 true（默认值是 false），免得线程抢锁呈现重大的不偏心，导致二阶段解决线程长时间抢不到锁。
transactionTimeOut 默认值 6 秒太短了，如果事务执行工夫超过 6 秒，就可能导致音讯失落。倡议改到 1 分钟左右。

生产者 client 也有一个注意事项，如果有多组 broker，并且是 2 正本（有 1 个 Slave），应该关上 retryAnotherBrokerWhenNotStoreOK，免得某个 Slave 呈现故障当前，大量音讯发送失败。

除了比拟一些惯例的监控伎俩以外，咱们开发了一个监控程序做分布式对账。能够发现咱们的集群以及咱们提供的 SDK 是否有异样。

具体做法是在每个 Broker 上都建设一个监控专用的 Topic，监控程序应用咱们本人提供的 SDK 框架来连贯集群（就像咱们的业务用户那样），监控生产者会给每个集群发送大量音讯。而后查看发送是否胜利：

发送胜利	胜利
	刷盘超时
	Slave 超时
	Slave 不可用
发送失败	具体错误码

生产者只对这些后果进行打点，不判断是否失常，具体到监控（或者演练）场景能够配置不同的报警规定。

消费者收到了音讯会通过 TCP 旁路 Ack 生产者，生产者这边会做分布式对账，将对账后果打点：

* 收到音讯
* 音讯失落（或超时未收到音讯）
* 反复收到音讯
* 音讯生成到最终生产的时间差
* Ack 生产者失败（由消费者打点）

同样监控程序只负责打点，报警规定可另外配置。

这套机制也能够用于分布式性能压测和故障演练。在做压测的时候，每个音讯都 Ack 的话，对生产者的内存压力很大，因为它收回去的音讯，须要在内存中保留一段时间（直到达到这个音讯的对账工夫），这段时间消费者 Ack 或者反复 Ack 都须要记录。所以咱们实现了按比例抽样对账的性能，开启当前只有须要对账的音讯才会在内存中保留一段时间。

顺便说一下，咱们做压测时，合格的规范是异步生产不失败、生产不提早、每一个音讯都不失落。这样做是为了保障压测时能给出更加精确的，可供线上零碎参考的性能数字，而不是制作现实条件，谋求一个大的数字。比方异步生产比同步生产更软弱（压测 client 如果同步生产，broker 抖动的时候，同步 client 会被梗塞导致发送速度升高，于是升高了 broker 压力，音讯发送不容易失败，然而会看到发送速率在稳定），更贴近生产环境的理论状况，咱们就抉择异步生产来评估。

# 性能优化

Broker 默认的参数在咱们的场景下（SSD、同步复制、异步刷盘）不是最优的，有的参数兴许在大多数场景下都不是最优的。咱们列出一些重要的参数，供大家参考：

参数	默认值	阐明
flushCommitLogTimed	False	默认值不合理，异步刷盘这个参数应该设置成 true，导致频繁刷盘，对性能影响极大
deleteWhen	04	几点删除过期文件的工夫，删除文件时有很多磁盘读，这个默认值是正当的，有条件的话还是倡议低峰删除
sendMessageThreadPoolNums	1	解决生产音讯的线程数，这个线程干的事件很多，倡议设置为 2～4，但太多也没有什么用。因为最终写 commit log 的时候只有一个线程能拿到锁。
useReentrantLockWhenPutMessage	False	如果前一个参数设置比拟大，这个最好设置为 True，防止高负载下自旋锁空转耗费 CPU。
sendThreadPoolQueueCapacity	10000	解决生产音讯的队列大小，默认值可能有点小，比方 5 万 TPS（异步发送）的状况下，卡 200ms 就会爆。设置比拟小的数字可能是放心有大量大音讯撑爆内存（比方 100K 的话，1 万个的音讯大略占用 1G 内存，也还好），具体能够本人算，如果都是小音讯，能够把这个数字改大。能够批改 broker 参数限度 client 发送大音讯。
brokerFastFailureEnable	True	Broker 端疾速失败（限流），和上面两个参数配合。这个机制可能有争议，client 设置了超时工夫，如果 client 还违心等，并且 sendThreadPoolQueue 还没有满，不应该失败，sendThreadPoolQueue 满了天然会回绝新的申请。但如果 client 设置的超时工夫很短，没有这个机制可能导致音讯反复。能够自行决定是否开启。现实状况下，能依据 client 设置的超时工夫来清理队列是最好的。
waitTimeMillsInSendQueue	200	200ms 很容易导致发送失败，倡议改大，比方 1000
osPageCacheBusyTimeOutMills	1000	Page cache 超时工夫，如果内存比拟多，比方 32G 以上，倡议改大点

得益于简略、简直 0 依赖的部署模式，使得咱们部署小集群的老本非常低；不对社区版本进行魔改，保障咱们能够及时降级；对立 SDK 入口不便集群保护和性能降级；通过复合小集群 + 主动负载平衡实现多机房多活；充分利用 RocketMQ 的性能比方事务音讯、提早音讯（加强）来满足业务的多样性需要；通过主动的分布式对账，对每一个 Broker 以及咱们的 SDK 进行正确性监控；本问也进行了一些性能参数的分享，但写的比较简单，根本只说了怎么调，但没能细说为什么，当前咱们会另写文章详述。目前 RocketMQ 曾经利用在公司在大多数业务线，期待未来会有更好的倒退！

扫码理解更多技术干货与客户案例：

*

> 版权申明： 本文内容由阿里云实名注册用户自发奉献，版权归原作者所有，阿里云开发者社区不领有其著作权，亦不承当相应法律责任。具体规定请查看《阿里云开发者社区用户服务协定》和《阿里云开发者社区知识产权爱护指引》。如果您发现本社区中有涉嫌剽窃的内容，填写侵权投诉表单进行举报，一经查实，本社区将立即删除涉嫌侵权内容。

正文完