本文作者：蒋兴涛，网易云音乐服务端开发工程师。云音乐有直播、解说、广告等多种线上场景。每个业务线都会有发彩票等消息场景，还有延时消息和交易消息场景，还有大数据做内嵌数据、数据清洗、离线处理等。云音乐在线RocketMQ主题10000+/天，峰值QPS流量150万/s，日消息量1000亿条。为了支撑庞大的数据规模和场景，除了搭建开源的RocketMQ集群外，我们还提升了监控和工具体验。监控提升主要包括监控整个集群的容量、状态、水位等健康状态，为消息的发送和消费提供流量、延迟、故障、耗时等监控指标。基于以上监控指标，还需要搭建业务巡检系统，实现在线报警。此外，我们还提供了一些修改工具来帮助业务方提升RocketMQ的使用体验，比如数据迁移和同步消息路由的组件、稳定性保障的限流能力、降级能力、动态参数干预的预规划能力等。.当线上业务方发现消费不符合预期时，需要提供查询帮助其快速定位，提供死信处理工具等。云音乐目前拥有三个机房，每个机房部署一个RocketMQ集群。除了Manesrv、HA等基础组件外，还有自研或开源改造的组件，如监控组件、告警巡检组件、降级维稳组件等。每个机房都有一套平台化的管控组件。管控端包括工单提交、在线离线、数据查询、订阅发布，以及一套消息通道平台和数据库。网易云音乐有多个流量入口，不同业务的数据和流量需要隔离。每个租户都有独立的业务线。物理隔离的成本太高，所以我们实现了逻辑隔离。各个业务之间的流量不互通，逻辑上不能相互调用，租户下的所有topic名称都是一样的。中台只需切换租户名称，无需更改任何配置或代码，即可直接上线。所有主题都在一个物理集群中，每个租户都有自己的一组逻辑集群，这些逻辑集群有自己的主题。不同逻辑集群中的Topic同名，实现了多租户隔离。随着云音乐的业务越来越大，业务方提出了更多的需求。比如异地多活，消息需要在多个机房消费，比如一般的埋点数据，多个产品的数据需要聚合到机场的数据处理集群进行离线处理，比如架构升级，不同单元之间的流量可以动态调度。基于以上需求，消息路由需要实现如下功能：①跨机房的消息复制。②流量去重：复制时消息路由难免会失败，所以必须有内部重试，可能导致消息重复；另外，双向路由必须提供双向复制，两边的topic名称相同，复制时会被复制。它会导致混乱，因此需要标签来实现大流量。③数据迁移任务。④监控完善，进度可控。云音乐的消息路由实现方案如上图所示。首先，在管控平台上会维护一套路由任务元数据表。业务方可以通过其他方式提交工单或申请路由任务，支持任意机房任意两个topic之间的消息路由。任务提交后，消息路由集群会周期性同步控制端消息路由任务的状态，同时向目标主题发送消息。路由任务可自行上报监控数据、消耗延时、累计监控报表等，可在管控端查看。云音乐的数据处理任务包括点埋和Trace，大部分使用Flink。但是由于开源方向并没有非常符合我们需求的connector，所以我们封装实现了自己的RocketMQFlinkconnector。由于内部封装了接口和集群配置，RocketMQ作为Flink的source和sink，需要进行数据源配置。我们封装数据源，比如连接器如何解析元数据，从而正确连接到数据源和读写消息。大数据任务的特点是测试环境和线上数据会混合在一起，多环境都有访问需求，所以我们设计了一套元数据，让connect能够连接多环境，处理多环境的流量指标和环境指标环境。和其他标签过滤。Flink有自己的检查点机制。只有在执行检查点时，消费者偏移量才会提交给代理。同时，需要管理消费者抵消。否则，消费站点消失会导致数据重新消费。因此，我们实现了状态管理机制。Flink的spottask比较敏感，抛出错误会导致task重新执行，连续重复几次后会导致TaskManagerfailover。另外，RocketMQ在网络场景中经常会出现brokerbusy或者网络问题，导致发送失败异常。我们为Flink定制了一套异常场景处理，使其不敏感。此外，我们目前面临的线上问题主要包括消息流量激增、机器负载高、突发大数据任务、重置消费位置等。当一个集群突然出现大流量时，其稳定性会受到很大影响，经常发送失败，其在线业务也会受到话题的影响。面对以上问题，除了提供隔离能力外，还需要限流和降级能力。首先，服务器的发送速率是有限的。支持主题级别和组级别。未来会支持client级别，支持多维度的sender限流。二是全球消费限流。分为topic和group，可以限制整个group消费关系下所有机器的总量，适用于大数据场景。三是单机消费有限。适用于在线业务场景，因为在线业务场景中每台机器的负载是有限的，不希望某个业务会无上限占用资源，所以需要限制单个的流量机器。当线上业务集群容量不够时，可以动态扩容增加容量。增加集群容量时无需修改全局容量。上图折线图反映了开启单机限流后，消费数据随着释放缓慢稳定增长，解决了流量尖峰，提高了在线集群的稳定性，稳定了消费服务的负载。随着集群规模的增大，消息延迟逐渐出现。经排查，发现producer可以正常向broker发送消息，但是由于数据量大，后台创建consumer队列的速度跟不上发送速度，导致消费延迟.其次，消费也面临瓶颈，跟不上发送速度，因为同一次发送可能有多个消费者。针对以上问题，我们进行了索引优化。在开源版本中，commitlog写入后，会有一套流程如Reput服务方法构建consumerqueue，index索引等，从头到尾，构建consumerqueue在块中。但是我们发现在保证顺序和站点顺序的前提下，索引是可以并发建立的，只要处理好站点的提交即可。因此，我们设置了报表队列的异步线程池，里面包含了不同的报表任务，每个任务都建立了自己的评论日志索引。一旦建立，索引不会立即可见。建立后，全局索引将被推回。如果之前的commitlog索引已经建好，后面的索引会立即可见，提高了创建索引的效率。上图是索引优化前后的性能对比。横轴代表主题数量，纵轴代表索引速度。灰色线代表索引速度，橙色线代表发送速度。优化前，主题较少时，索引速度比发送速度慢。随着题目数量的增加，两者的速度逐渐趋于一致，但性能明显下降。优化后，索引速度与发送速度基本一致，性能不会随着主题数量的增加而明显下降。优化后建索引性能是优化前的3倍，保证了消费性能。在广播消费场景中，通常需要保证所有机器上的数据最终一致。但是开源广播消费失败后不会重试，也不会报警。并且消费地点在本地，不会上报给远端。如果重启本地服务，偏移量将丢失，无法进行预先计划的干预。同时，由于不报告，缺乏定位问题的能力。我们的方案是逻辑组+实际组。逻辑组是指业务方在代码和管控平台中申请的消费组。申请群组后，在客户端设置，标识为新版本广播消费的群组。当每个实例启动时，在逻辑组中添加扩展以生成实际组。实际组可以进行正常的集群消费，也可以复用集群消费的所有能力，包括租户隔离、消息路由、监控、限流能力等，最终解决了广播消费的问题，可以使用死信和重试，重置消费位置，位置查询，监控报警功能。此外，云音乐还遇到了在线流量突然增加来不及发布，或者在线流量与预期不符，发布成本过高等问题。为此，我们为业务方提供了线程实时调整的能力。在开源版本中，每个client都会向broker上报ConsumerRunningInfo，包括消费是否暂停、订阅状态、订阅了哪些主题、消费位置等。我们添加了线程池的coresize和maxsize每个topic消耗到上报的信息，并显示在控制端，方便用户实时感知线程池的当前状态。此外，我们提供了一种非常简单的修改方式。在control端修改kvconfig，上报给NameSvr，NameSvr监控下发。客户端监听到kvconfig的变化后，拉取最新的配置。然后在本地找到题目对应的线程池，修改coresize和maxsize的值。此前，业务在线发现问题后往往需要10-20分钟才能启动。但是现在只需要修改一个参数，发送，轮询就可以了。整个过程不超过30秒。RocketMQ对云音乐的未来规划如下：第一，云原生。云原生具备弹性伸缩能力，可以更好的节约成本，应对线上突发风险。二是提高效率。比如主题签名迁移、从一个集群迁移到另一个集群、从顺序消息变为非顺序消息等操作，目前还没有完全白化。未来我们会在这方面提高效率和用户体验。第三，开源社区交流贡献。
