RocketMQ 在网易云音乐的实践

本文作者：蒋星韬，网易云音乐服务端开发工程师。

云音乐线上场景众多，比如直播、评论、广告，各个业务线都会有消息场景比如发奖券，也会有延迟消息和事务消息场景，以及大数据做埋点数据、数据清洗、离线处理等。

云音乐线上 RocketMQ topic 为 1 万+/天，QPS 流量峰值为 150 万/s，日消息量千亿级别。为了支撑庞大的数据规模和场景，除了搭建开源 RocketMQ 集群，我们也做了监控的完善和工具体验。监控完善主要包括对整个集群的容量、状态、水位进行健康状态的监控，针对消息的发送和消费提供流量、延迟、失败、耗时等监控指标。基于以上监控指标，还需搭建一套业务巡检体系，以实现线上告警。

另外，我们也提供改了一些工具帮助业务方提升使用 RocketMQ 的体验，比如数据迁移和同步消息路由的组件，提供稳定性保障的限流能力、降级能力以及动态参数干预的预案能力。当线上业务方发现消费不符合预期时，需要提供查询帮助其快速定位，以及提供死信处理工具等。

云音乐目前有三个机房，每个机房部署了一套 RocketMQ 集群，除了 Manesrv、 HA 等基础组件，还有自研或开源改造的组件，比如 monitor 组件、告警巡检组件、降级维稳组件等。

每个机房里有一套平台化的管控组件，管控端包含提工单、上下线、查数据、订阅问题，还包括一套消息路平台和数据库。

网易云音乐拥有多个流量入口，不同业务的数据和流量需要做隔离，每个租户下都是一套独立的业务线。而物理隔离成本过高，因此我们实现了逻辑隔离。各个业务之间流量不互通，逻辑上无法相互调用，且租户下所有 topic 名字一致，中台只需要切换租户名，无需改动任何其配置、代码，即可直接上线。

所有 topic 都在一个物理集群内，每个租户有自己的一套逻辑集群，逻辑集群内有自己的 topic，不同逻辑集群之间的 topic 同名，实现了多租户隔离。

随着云音乐的业务愈发庞大，业务方提出了更多需求。比如异地多活，消息需要在多个机房消费，比如通用埋点数据，需要将多个产品的数据汇总到机场的数据处理集群做离线处理，比如架构升级，不同单元间的流量能够动态调度。

基于以上需求，消息路由需要实现以下几个功能：

①跨机房消息复制。

②流量去重：消息路由在复制时不可避免会有失败，因此必然有内部的重试，可能会导致有消息重复；此外，双向路由必然需要提供双向复制，而两边 topic 名字一样，复制时会导致错乱，因此需要有标签来实现流量重。

③数据迁移任务。

④监控完善，进度可控。

云音乐的消息路由实现方案如上图所示。

首先，在管控平台会维护一套路由任务元数据表，业务方可以提工单或者通过其他方式申请路由任务，支持任意机房的任意两个 topic 之间做消息路由。任务提交之后，消息路由集群会定时同步管控端上的消息路由任务的状态，同时将消息发送到目标 topic 。路由任务能够自行上报监控数据、消费延迟、堆积监控报表等，可在管控端进行查看。

云音乐的数据处理任务包括埋点、trace，大多使用 Flink。但由于开源方向没有与我们的需求非常匹配的 connector ，因此我们封装实现了自己的 RocketMQ Flink connector。

因为内部封装了接口和集群配置，RocketMQ 作为 Flink 的 source 和 sink 需要有数据源的配置。我们对数据源做了封装，比如 connector 如何解析元数据，从而正确地连接数据源、读写消息。

大数据任务的特点为测试环境与线上数据会混在一起，多环境都有接入需求，因此我们设计了一套元数据，使得 connect 能够连接多环境且能够处理多环境里面流量标、环境标等标签的过滤。

Flink 有自己的 checkpoint 机制，只有在做 checkpoint 时才会将 consumer offset 提交给 broker ，同时需要对 consumer offset 进行管理，否则消费位点消失会导致数据重新消费，因此我们实现了 state 管理机制。

Flink 的 spot task 比较敏感，抛出错误则会导致 task 重新执行，连续重复几次后会导致 TaskManager failover 。此外，RocketMQ 在网络场景下时常出现 broker busy 或网络问题导致发送失败异常。我们针对 Flink 定制了一套异常场景处理，使其变得不敏感。

此外，我们目前面临的线上问题主要包括消息流量激增、机器负载高、大数据任务突刺、重置消费位点等。集群突然出现大流量行动时，其稳定性会受到极大冲击，频繁发送失败，线上其业务也会受到 topic 的影响。

面对以上问题，除了提供隔离能力外，也需要限流降级的能力。

第一，服务端的发送限流。支持 topic 级别，也支持 group 级别。后续将支持客户端级别，支持多个维度的发送端限流。

第二，全局消费限流。分为 topic 和 group，可以对整个 group 消费关系下所有机器的总量进行限流，适用于大数据场景。

第三，单机消费限流。适用于线业务场景，因为在线业务场景每台机器的负载有限，不希望某个业务无上限地占用资源，因此需要对单机限流。在线业务集群容量不够时，可以做动态扩容来增加容量。增加集群容量时无需修改全局容量。

上图折线图反应了开启单机限流之后，消费数据随着发布缓慢平稳上涨，解决了流量突刺，提升线上集群稳定性和消费服务的负载平稳。

随着集群规模增大，逐渐出现了消息延迟的情况。经排查发现，producer 能够正常发送消息到 broker ，但是由于数据量非常大，后台创建 consumer queue 的速度跟不上发送速度，导致消费延迟。其次，消费也面临瓶颈，跟不上发送速度，因为同一个发送可能存在多个消费方。

针对以上问题，我们进行了索引优化。

开源版本下， commit log 写入之后，会有 Reput service 方法建 consumer queue 、index 索引等一套流程，从头扫到尾，块状地建立 consumer queue。

而我们发现，保证顺序性和位点的有序性的前提下，可以并发地建索引，只需处理好位点的提交即可。因此，我们设置了 reput queue 异步线程池，里面有不同的 reput task ，每个 task 建立自己的 comment log 索引。建好之后，索引并不是立刻可见。建好之后会有全局的索引往后推，如果前面的 commit log 索引已经建好，则后面的索引也立马可见，提升了索引的建立效率。

上图为索引优化前后的性能对比。横轴代表 topic 数量，纵轴代表建索引的速度。灰色线代表建索引的速度，橙色线代表发送速度。

优化前，topic 较少时，建索引的速度慢于发送速度。随着 topic 数量增多，两者速度逐渐一致，但性能均明显下降。

优化后，建索引的速度基本与发送速度持平，且性能不会随着 topic 数量增加而大幅下降。优化后建索引的性能达到优化前的 3 倍，保证了消费性能。

广播消费场景下，通常需要确保所有机器上的数据最终一致。而开源的广播消费失败后不会重试也不会告警。且消费位点为 local，不会上报到远端，如果本地服务重启则 offset 丢失，并且无法做预案干预。同时因为不上报，缺少问题定位的能力。

我们的解决方案为逻辑 group +实际 group 。

逻辑 group 指业务方在代码和在管控平台申请的 consumer group 。申请 group 之后，在客户端进行设定，将其标识为新版广播消费的 group，每个实例启动时在逻辑 group 后加上扩展名来生成实际 group 。

实际 group 可以进行正常的集群消费，也可以用复用集群消费的所有能力，包括租户隔离、消息路由、监控、限流能力等，最终就解决了广播消费的问题，能够使用死信、重试、重置消费位点、位点查询、监控告警等能力。

此外，云音乐日常还会出现线上流量突增来不及发布，或线上流量与预期不符而发布成本过高等问题。

为此，我们为业务方提供了实时线程式调整的能力。开源版本中，每个客户端都会向 broker 做 Consumer RunningInfo 的上报，包括是否消费暂停、subscribe 的状态、订阅了哪些 topic、消费位点等。我们在在上报信息里加上了每个 topic 自己消费的线程池的 coresize、maxsize ，并在管控端展示，使用户能够实时感知当前线程池的状态。

此外，我们提供了非常简单的修改方式。在管控端修改 kv config 并上报到 NameSvr ，NameSvr 监听并下发。客户端监听到 kv config 变化后将最新配置下拉。然后再本地找到 topic 对应的线程池，修改 coresize、maxsize 值。

此前，业务在线上发现问题后发布往往需要 10-20 分钟起步。而现在只需修改一个参数、下发、轮询即可完成，整个过程不超过 30 秒。

云音乐在 RocketMQ 的未来规划如下：

第一，云原生。云原生有弹性扩缩容的能力，可以更好地节约成本以及应对线上突发风险。

第二，提效率。比如 Topic 签迁移、从一个集群迁移到另一个集群、从顺序消息改为非顺序消息等操作目前还未实现完全白屏化，后续会针对此方面提高效率，提高用户体验。

第三，开源社区交流贡献。