mysql基于binlake同步ES积压解决方案-京东云开发者社区

### 1 背景与目标

#### 1.1 背景

国际财务泰国每月月初账单任务生成，或者重算账单数据，数据同步方案为mysql通过binlake同步ES数据，在同步过程中发现计费事件表，计费结果表均有延迟，ES数据与Mysql数据不一致，导致业务页面查询数据不准确，部分核心计算通过ES校验失败

#### 1.2目标

解决binlake到JMQ积压同步ES延迟问题

### 2 当前业务流程

#### 2.1 流程图

现有业务基本流程如下图，包含运营端和外部数据接入，整体操作到数据存储流程

![](//img1.jcloudcs.com/developer.jdcloud.com/f9dbf0a6-9032-4dc0-9f4a-f4fa79f740f620221012152230.png)

#### 2.2 数据流

![](//img1.jcloudcs.com/developer.jdcloud.com/1c01e3d4-5d1f-411a-b749-55f0b35e4ef820221012152321.png)

### 3 问题分析

#### 3.1 问题现象

jmq积压，报警
国内站截图如下

![](//img1.jcloudcs.com/developer.jdcloud.com/02c47c0c-2b4a-48c0-bdc1-c575769fdc7620221012152408.png)

![](//img1.jcloudcs.com/developer.jdcloud.com/50ddf2a2-c85f-40bd-9871-ce7da154022820221012152421.png)

#### 3.2 筛查分析

普及：JMQ默认生产者发送消息QPS受到主题的broker数量影响，(8w/s)/broker

##### 3.2.1 MQ积压分析

1）分析原因一、ES写入量大，导致ES写入QPS瓶颈

ES写入瓶颈需要进行压测，才能确定实际是否达到瓶颈；
通过查询集群负载，写入队列有无积压，cpu高不高，来定位
以下为调整MQ批量消费大小后的ES监控
写入队列无积压，CPU不高，写入QPS没有达到瓶颈

![](//img1.jcloudcs.com/developer.jdcloud.com/b3ccf992-79bc-406d-9c21-ea2880bb122320221012152605.jpg)

![](//img1.jcloudcs.com/developer.jdcloud.com/acdbea99-015d-4263-b25f-fec4e8b5a53920221012152629.png)

2）分析原因二、ES写入慢导致消费积压

ES解析服务解析慢，瓶颈在ES解析处
根据当前系统CPU、负载信息定位是否服务器性能满负荷，是否扩容
无报警信息，整体运行平稳，基本排除业务资源达到瓶颈问题引起写入慢

![](//img1.jcloudcs.com/developer.jdcloud.com/a5078080-5ac8-4bc4-aa78-365d6f6e748420221012152702.png)

MQ消费端消费慢，瓶颈在消费并发处
当前主题分片数3，队列数为15，默认最大并发数为15*10，报警当时入队数500~700/s
定位问题，为MQ消费慢，其根本原因为受到ES-Parse业务系统处理速度影响

#### 3.3 临时处理方案

开启mq并行消费策略，写入QPS显著增加

![](//img1.jcloudcs.com/developer.jdcloud.com/cffc80d7-f99e-40da-a346-223c1f89091e20221012152748.png)

### 4 如何提升消费速率，提升写入ES速率

造成问题原因核心点是MQ积压，业务系统消费慢，MQ入队数大于出队数，导致积压

#### 4.1 原理分析

##### 4.1.1 存储流程解析

第一步：binlake订阅mysql binlog
第二步：发MQ，JMQ数据传输
第三步：消费JMQ数据，ES Paser数据解析，
第四步：数据存储

![](//img1.jcloudcs.com/developer.jdcloud.com/0fe810d5-9b83-4665-8061-411ebaa984e720221012152825.png)

##### 4.1.2 binlake基本原理

![](//img1.jcloudcs.com/developer.jdcloud.com/4c21b7a2-6b50-493b-b76a-08a8199ab6f620221012152851.png)

##### 4.1.3 binlake发送MQ过程

![](//img1.jcloudcs.com/developer.jdcloud.com/7263b7f7-a4a8-407a-976e-2dc5d7c206e420221012152932.png)

##### 4.1.4 JMQ消费原理

JMQ消费默认就是批量消费
消费原理如下图

![](//img1.jcloudcs.com/developer.jdcloud.com/7b883120-5de3-4bbd-b482-279607f8ff6520221012152959.png)

批量消费与并行消费原理如下图

![](//img1.jcloudcs.com/developer.jdcloud.com/82d45671-5d17-4c5f-827f-7adfd78e729e20221012153023.png)

通过分析，在未开启并行消费前提下，当前主题最大处并发的消费处理能力即是队列数

#### 4.2 提升消费速率的几种方案

##### 4.2.1MQ增加消费速度方法

**扩容，增加并发消费能力**
针对MQ默认情况下，一切扩容都能解决问题，增大分片数，增加队列数
需要额外资源，申请扩容新的broker，同时考虑增加消费端实例

**增加批量大小**
首先保证，业务系统(ES-Parse)消费MQ消息，处理10条和处理100条速度基本一样
实践：国际财务针对此方法进行代码逻辑改造

**开启并行数**
理论上增加（并行数/批量数）的倍数并发处理能力
要求数据无序，针对乱序，数据存储，不影响业务

##### 4.2.2 并行有序的方案

1）实现数据幂等性，增加缓存，并行消费策略

方案流程

![](//img1.jcloudcs.com/developer.jdcloud.com/12bc6dda-a6b3-44d0-a862-60d07daf423c20221012153143.png)

基础实现流程：

1）根据binlake发送mq，在mq端开启并行消费，确保并行消费
2）根据业务单号对，单号加锁（如麦哲伦对运单号加锁，即对单号加分布式锁），根据对应的ID获取ES数据。
3）校验数据是否有效，若查询无数据，则直接新增；若查询的数据状态大于当前数据状态，则直接抛弃，若查询状态小于当前数据状态，则直接更新数据
4）更新缓存并释放锁

优点

- 指定资源情况下，增大消费端并发
- 可以开启并行消费，且保证顺序消费
- 可以使得资源充分利用，增加消费性能

缺点

- 增加毫秒级缓存额外开销

实践：麦哲伦运单中心针对此方案实现binlake数据同步ES

2）binlake主题分发子主题，显示增大并发策略

![](//img1.jcloudcs.com/developer.jdcloud.com/31eab3b7-9b00-444c-a2c7-990a685e7e9620221012153235.png)

优点：

- 逻辑相对简单，不需要开发复杂逻辑，无需引入额外中间件
- 预估转发消息速率即是实际处理速率

提升速率计算:

- 原主题单线程处理一条数据存储到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进行分析，同时针对积压推荐俩种方案，并从性能合理利用及扩展性分析，简要介绍方案二并行有序消费策略，希望能够帮助大家，如有问题，请随时指出！

------------
###### 自猿其说Tech-JDL京东物流技术与数据智能部
###### 作者：任洪波