1 背景与目标
1.1 背景
国际财务泰国每月月初账单任务生成，或者重算账单数据，数据同步方案为 mysql 通过 binlake 同步 ES 数据，在同步过程中发现计费事件表，计费结果表均有延迟，ES 数据与 Mysql 数据不一致，导致业务页面查询数据不准确，部分核心计算通过 ES 校验失败
1.2 目标
解决 binlake 到 JMQ 积压同步 ES 延迟问题
2 当前业务流程
2.1 流程图
现有业务基本流程如下图，包含运营端和外部数据接入，整体操作到数据存储流程
2.2 数据流
3 问题分析
3.1 问题现象
jmq 积压，报警
国内站截图如下
3.2 筛查分析
普及：JMQ 默认生产者发送消息 QPS 受到主题的 broker 数量影响，(8w/s)/broker
3.2.1 MQ 积压分析
1）分析原因一、ES 写入量大，导致 ES 写入 QPS 瓶颈
ES 写入瓶颈需要进行压测，才能确定实际是否达到瓶颈；
通过查询集群负载，写入队列有无积压，cpu 高不高，来定位
以下为调整 MQ 批量消费大小后的 ES 监控
写入队列无积压，CPU 不高，写入 QPS 没有达到瓶颈
2）分析原因二、ES 写入慢导致消费积压
ES 解析服务解析慢，瓶颈在 ES 解析处
根据当前系统 CPU、负载信息定位是否服务器性能满负荷，是否扩容
无报警信息，整体运行平稳，基本排除业务资源达到瓶颈问题引起写入慢
MQ 消费端消费慢，瓶颈在消费并发处
当前主题分片数 3，队列数为 15，默认最大并发数为 15*10，报警当时入队数 500~700/s
定位问题，为 MQ 消费慢，其根本原因为受到 ES-Parse 业务系统处理速度影响
3.3 临时处理方案
开启 mq 并行消费策略，写入 QPS 显著增加
4 如何提升消费速率，提升写入 ES 速率
造成问题原因核心点是 MQ 积压，业务系统消费慢，MQ 入队数大于出队数，导致积压
4.1 原理分析
4.1.1 存储流程解析
第一步：binlake 订阅 mysql binlog
第二步：发 MQ，JMQ 数据传输
第三步：消费 JMQ 数据，ES Paser 数据解析，
第四步：数据存储
4.1.2 binlake 基本原理
4.1.3 binlake 发送 MQ 过程
4.1.4 JMQ 消费原理
JMQ 消费默认就是批量消费
消费原理如下图
批量消费与并行消费原理如下图
通过分析，在未开启并行消费前提下，当前主题最大处并发的消费处理能力即是队列数
4.2 提升消费速率的几种方案
4.2.1MQ 增加消费速度方法
扩容，增加并发消费能力
针对 MQ 默认情况下，一切扩容都能解决问题，增大分片数，增加队列数
需要额外资源，申请扩容新的 broker，同时考虑增加消费端实例
增加批量大小
首先保证，业务系统 (ES-Parse) 消费 MQ 消息，处理 10 条和处理 100 条速度基本一样
实践：国际财务针对此方法进行代码逻辑改造
开启并行数
理论上增加（并行数/批量数）的倍数并发处理能力
要求数据无序，针对乱序，数据存储，不影响业务
4.2.2 并行有序的方案
1）实现数据幂等性，增加缓存，并行消费策略
方案流程
基础实现流程：
1）根据 binlake 发送 mq，在 mq 端开启并行消费，确保并行消费
2）根据业务单号对，单号加锁（如麦哲伦对运单号加锁，即对单号加分布式锁），根据对应的 ID 获取 ES 数据。
3）校验数据是否有效，若查询无数据，则直接新增；若查询的数据状态大于当前数据状态，则直接抛弃，若查询状态小于当前数据状态，则直接更新数据
4）更新缓存并释放锁
优点
指定资源情况下，增大消费端并发
可以开启并行消费，且保证顺序消费
可以使得资源充分利用，增加消费性能
缺点
增加毫秒级缓存额外开销
实践：麦哲伦运单中心针对此方案实现 binlake 数据同步 ES
2）binlake 主题分发子主题，显示增大并发策略
优点：
逻辑相对简单，不需要开发复杂逻辑，无需引入额外中间件
预估转发消息速率即是实际处理速率
提升速率计算:
原主题单线程处理一条数据存储到 ES 时间为 es_time，举例为 50ms，每秒吞吐量是 20 条
现单线程转发 MQ 一条数据时间为 trans_time，举例为 20ms，每秒转发吞吐量 50 条
假设转发 topic 为 N 个子主题，则吞吐量理论为 n*20 实际小于转发吞吐量 50，此处多子主题对 cpu 核数竞争
提升吞吐量为=(1000ms/trans_time ) 转发吞吐量 - (1000ms/es_time) 原有吞吐量
缺点
扩展性不好，实际结果有待验证，小于预估值
实践：跨境赤道分发中心实现类似功能实践，消息转发，其他 MQ 实现
3）俩种方案对比
主题较少一个俩个主题情况下，且业务处理比较耗时情况下，不想额外开发，可选方案二
长期方案选择方案一，并行消费策略，可伸缩性，可扩展，支持动态扩容
5.总结
针对 MQ 积压问题，并行消费可以是解决问题的一大利器，本文从 binlake 同步 ES 进行分析，同时针对积压推荐俩种方案，并从性能合理利用及扩展性分析，简要介绍方案二并行有序消费策略，希望能够帮助大家，如有问题，请随时指出！
作者：任洪波