本文作者：蒋星韬，网易云音乐服务端开发工程师。
云音乐线上场景众多，比如直播、评论、广告，各个业务线都会有消息场景比如发奖券，也会有延迟消息和事务消息场景，以及大数据做埋点数据、数据清洗、离线处理等。
云音乐线上 RocketMQ topic 为 1 万+/天，QPS 流量峰值为150万/s，日消息量千亿级别。为了 694 支撑庞大的 311 数据规模和场景，除了 75 搭建开源RocketMQ集群，我 28 们也做了 1451 监控的 20 完善和工具体验。监控完善主要包括对整个集群的 1600 容量、状态、水位进行健康状态的 1267 监控，针对消息的 877 发送和消费提供流量、延迟、失败、耗时等监控指标。基于以上监控指标，还需搭建一套业务巡检体系，以实现线上告警。
另外，我 648 们也提供改了 630 一些工具帮助业务方提升使用 RocketMQ 的 1226 体验，比如数据迁移和同步消息路由的 52 组件，提供稳定性保障的 625 限流能力、降级能力以及动态参数干预的 1528 预案能力。当线上业务方发现消费不符合预期时，需要提供查询帮助其快速定位，以及提供死信处理工具等。
云音乐目前有三个机房，每个机房部署了 262 一套 RocketMQ 集群，除了 823 Manesrv、 HA 等基础组件，还有自研或开源改造的 264 组件，比如 monitor 组件、告警巡检组件、降级维稳组件等。
每个机房里有一套平台化的 391 管控组件，管控端包含提工单、上下线、查数据、订阅问题，还包括一套消息路平台和数据库。
网易云音乐拥有多个流量入口，不同业务的 1388 数据和流量需要做隔离，每个租户下都是 1553 一套独立的 1252 业务线。而物理隔离成本过高，因此我 1413 们实现了 1058 逻辑隔离。各个业务之间流量不互通，逻辑上无法相互调用，且租户下所有 topic 名字一致，中台只需要切换租户名，无需改动任何其配置、代码，即可直接上线。
所有 topic 都在一个物理集群内，每个租户有自己的 1140 一套逻辑集群，逻辑集群内有自己的 765 topic，不同逻辑集群之间的 734 topic 同名，实现了 1168 多租户隔离。
随着云音乐的 1218 业务愈发庞大，业务方提出了 1308 更多需求。比如异地多活，消息需要在多个机房消费，比如通用埋点数据，需要将多个产品的 197 数据汇总到机场的 1251 数据处理集群做离线处理，比如架构升级，不同单元间的 861 流量能够动态调度。
基于以上需求，消息路由需要实现以下几个功能：
①跨机房消息复制。
②流量去重：消息路由在复制时不可避免会有失败，因此必然有内部的 1480 重试，可能会导致有消息重复；此外，双向路由必然需要提供双向复制，而两边 topic 名字一样，复制时会导致错乱，因此需要有标签来实现流量重。
③数据迁移任务。
④监控完善，进度可控。
云音乐的 1379 消息路由实现方案如上图所示。
首先，在管控平台会维护一套路由任务元数据表，业务方可以提工单或者通过其他 513 方式申请路由任务，支持任意机房的 534 任意两个 topic 之间做消息路由。任务提交之后，消息路由集群会定时同步管控端上的 588 消息路由任务的 322 状态，同时将消息发送到目标 topic 。路由任务能够自行上报监控数据、消费延迟、堆积监控报表等，可在管控端进行查看。
云音乐的 1250 数据处理任务包括埋点、trace，大多使用Flink。但由于开源方向没有与我 857 们的 164 需求非常匹配的 442 connector ，因此我 875 们封装实现了 1498 自己的 272 RocketMQ Flink connector。
因为内部封装了 502 接口和集群配置，RocketMQ 作为 Flink 的 1217 source 和 sink 需要有数据源的 1494 配置。我 11 们对数据源做了 1572 封装，比如 connector 如何解析元数据，从而正确地连接数据源、读写消息。
大数据任务的 117 特点为测试环境与线上数据会混在一起，多环境都有接入需求，因此我 1395 们设计了 78 一套元数据，使得 connect 能够连接多环境且能够处理多环境里面流量标、环境标等标签的 1368 过滤。
Flink有自己的 1215 checkpoint 机制，只有在做 checkpoint 时才会将 consumer offset 提交给 broker ，同时需要对 consumer offset 进行管理，否则消费位点消失会导致数据重新消费，因此我 911 们实现了 866 state 管理机制。
Flink的 140 spot task 比较敏感，抛出错误则会导致 task 重新执行，连续重复几次后会导致TaskManager failover 。此外，RocketMQ 在网络场景下时常出现broker busy 或网络问题导致发送失败异常。我 907 们针对Flink 定制了 1380 一套异常场景处理，使其变得不敏感。
此外，我 665 们目前面临的 643 线上问题主要包括消息流量激增、机器负载高、大数据任务突刺、重置消费位点等。集群突然出现大流量行动时，其稳定性会受到极大冲击，频繁发送失败，线上其业务也会受到 topic 的 1322 影响。
面对以上问题，除了 731 提供隔离能力外，也需要限流降级的 369 能力。
第一，服务端的 1152 发送限流。支持 topic 级别，也支持 group 级别。后续将支持客户端级别，支持多个维度的 572 发送端限流。
第二，全局消费限流。分为 topic 和 group，可以对整个 group 消费关系下所有机器的 680 总量进行限流，适用于大数据场景。
第三，单机消费限流。适用于线业务场景，因为在线业务场景每台机器的 228 负载有限，不希望某个业务无上限地占用资源，因此需要对单机限流。在线业务集群容量不够时，可以做动态扩容来增加容量。增加集群容量时无需修改全局容量。
上图折线图反应了 733 开启单机限流之后，消费数据随着发布缓慢平稳上涨，解决了 389 流量突刺，提升线上集群稳定性和消费服务的 154 负载平稳。
随着集群规模增大，逐渐出现了 1224 消息延迟的 87 情况。经排查发现，producer 能够正常发送消息到 broker ，但是 1167 由于数据量非常大，后台创建 consumer queue 的 210 速度跟不上发送速度，导致消费延迟。其次，消费也面临瓶颈，跟不上发送速度，因为同一个发送可能存在多个消费方。
开源版本下， commit log 写入之后，会有 Reput service 方法建 consumer queue 、index 索引等一套流程，从头扫到尾，块状地建立 consumer queue。
而我 1133 们发现，保证顺序性和位点的 35 有序性的 1057 前提下，可以并发地建索引，只需处理好位点的 868 提交即可。因此，我 1595 们设置了 968 reput queue 异步线程池，里面有不同的 527 reput task ，每个 task 建立自己的 141 comment log 索引。建好之后，索引并不是 688 立刻可见。建好之后会有全局的 546 索引往后推，如果前面的 863 commit log 索引已经建好，则后面的 1449 索引也立马可见，提升了 789 索引的 81 建立效率。
上图为索引优化前后的 1569 性能对比。横轴代表 topic 数量，纵轴代表建索引的 779 速度。灰色线代表建索引的 1518 速度，橙色线代表发送速度。
优化前，topic 较少时，建索引的 814 速度慢于发送速度。随着 topic 数量增多，两者速度逐渐一致，但性能均明显下降。
优化后，建索引的 1424 速度基本与发送速度持平，且性能不会随着 topic 数量增加而大幅下降。优化后建索引的 948 性能达到优化前的 97 3倍，保证了 969 消费性能。
广播消费场景下，通常需要确保所有机器上的 1011 数据最终一致。而开源的 905 广播消费失败后不会重试也不会告警。且消费位点为 local，不会上报到远端，如果本地服务重启则offset 丢失，并且无法做预案干预。同时因为不上报，缺少问题定位的 729 能力。
我 691 们的 1441 解决方案为逻辑 group +实际 group 。
逻辑 group 指业务方在代码和在管控平台申请的 1128 consumer group 。申请 group 之后，在客户端进行设定，将其标识为新版广播消费的 346 group，每个实例启动时在逻辑group 后加上扩展名来生成实际 group 。
实际 group 可以进行正常的 880 集群消费，也可以用复用集群消费的 1609 所有能力，包括租户隔离、消息路由、监控、限流能力等，最终就解决了 201 广播消费的 313 问题，能够使用死信、重试、重置消费位点、位点查询、监控告警等能力。
此外，云音乐日常还会出现线上流量突增来不及发布，或线上流量与预期不符而发布成本过高等问题。
为此，我 1258 们为业务方提供了 340 实时线程式调整的 1508 能力。开源版本中，每个客户端都会向broker 做 Consumer RunningInfo 的 279 上报，包括是 1355 否消费暂停、subscribe 的 1360 状态、订阅了 448 哪些 topic、消费位点等。我 1425 们在在上报信息里加上了 662 每个 topic 自己消费的 954 线程池的 1339 coresize、maxsize ，并在管控端展示，使用户能够实时感知当前线程池的 644 状态。
此外，我 714 们提供了 1141 非常简单的 1134 修改方式。在管控端修改 kv config 并上报到 NameSvr ，NameSvr 监听并下发。客户端监听到 kv config 变化后将最新配置下拉。然后再本地找到 topic 对应的 1580 线程池，修改 coresize、maxsize 值。
此前，业务在线上发现问题后发布往往需要 10-20 分钟起步。而现在只需修改一个参数、下发、轮询即可完成，整个过程不超过 30 秒。
云音乐在 RocketMQ 的 1018 未来规划如下：
第一，云原生。云原生有弹性扩缩容的 449 能力，可以更好地节约成本以及应对线上突发风险。
第二，提效率。比如 Topic 签迁移、从一个集群迁移到另一个集群、从顺序消息改为非顺序消息等操作目前还未实现完全白屏化，后续会针对此方面提高效率，提高用户体验。
第三，开源社区交流贡献。