这个技术方案源于我项目开发中遇到的一个技术问题：数据同步大宽表。那么啥是数据同步大宽表勒？数据同步大宽表，顾名思义就是把多张mysql中的业务数据表组合成一张大宽表，一般业内选择使用es来存储大宽表。

为什么要用ES构建大宽表

背景1：mysql多表join性能很差

举个例子：订单系统涉及到了三张表商品表记录商品信息，订单主表记载了所有订单主单的信息，一个订单可能有好几个子单构成，所以需要订单子单表。这就是三张表了！那么这个时候你要查询一个订单的详情信息，这个订单的详情信息要查这三张表才能凑齐，请问怎么保证又快又高效的查出来。

如果你要使用mysql解决这个问题，那么很显然只能多表join。但是真实的企业级开发中，一个表数据量都在上千万行，表与表的join本质上是做笛卡尔积，时间复杂度很高，阿里巴巴规约也明确表示超过三张表禁止使用join，所以很显然为了查询性能我们不能贸然用多表join完成查询。这个时候大宽表的优势就凸显出来了，这张大宽表的信息是很全的，我们直接查就行，不存在join！

背景2: mysql无法应对复杂条件查询

有人可能会问：我们可以在mysql中建一个大宽表呀，为什么要用es勒？
因为MySQL 架构天生不适合海量数据查询（一千万行以内），无法应对海量数据下各种复杂条件的查询。有人说建索引呀！加索引确实可以提升查询速度，但在 MySQL 中加多个索引最终在执行 SQL 的时候它只会选择成本最低的那个索引。如果没有索引满足搜索条件，就会触发全表扫描，而且即便你使用了组合索引，也要符合最左前缀原则才能命中索引。但在海量数据多种查询条件下很有可能不符合最左前缀原则而导致索引失效。
此外每加一个索引就会创建一颗B+树，如果是海量数据那将大大增加存储成本，所以索引绝对不是越多越好的！
与其说上面列的这些点是 MySQL 的不足，倒不如说 MySQL 本身就不是为海量数据查询而设计的。术业有专攻，海量数据查询还得用专门的搜索引擎，这其中 ES 是当之无愧的王者。它有如下三个特点：

每个字段都被索引（ES中一切字段皆为索引），而且采用的是高效的倒排索引
有丰富的分词插件，支持自定义打分和排序，能够实现任意复杂条件的全文检索
支持分布式存储，轻松处理 PB 级别的结构化或非结构化数据。高可用，容灾性能好。

背景3：mysql模糊查询和全文查询能力弱

有些查询条件是 MySQL 加索引都解决不了的，比如我要查询商品中所有 title 带有「格力空调」的关键词，如果你用 MySQL 写，会写出如下代码：

SELECT * FROM product WHERE title like '%格力空调%'

这样的话无法命中任何索引，会触发全表扫描，而且你不能指望所有人都能输对他想要的商品，是人就会犯错误，我们经常会犯类似把「格力空调」记成「格空调」的错误，但是我还是希望要找到服务能够格力空调。那么 SQL 语句就会变成：

SELECT * FROM product WHERE title like '%格空调%'

这种情况下就算你触发了全表扫描也无法查询到任何商品，综上所述，MySQL 的查询确实能力有限。
ES本身就是全文搜索引擎，有丰富的分词插件，支持自定义打分和排序，甚至可以引入检索算法实现语义理解！

在这种特殊的查询诉求下，我们只能求助于ES，但是平时的业务数据还是写进mysql里面呀，我们怎么把业务数据库里的数据同步到ElasticSearch中嘞？

2. 怎么完成MySQL到ES的数据同步

目前主要有四种方案：同步双写、异步双写、定时任务调度、binlog同步

2.1 同步双写

这是一种最为简单的方式，在将数据写到mysql时，同时将数据写到ES。
优点：

业务逻辑很简单

缺点：

硬编码，有需要写入mysql的地方都需要添加写入ES的代码；
业务强耦合
存在双写失败丢数据风险；
性能较差：本来mysql的性能不是很高，再加一个ES，系统的性能必然会下降。

针对这种情况，有数据强一致性要求的，就必须双写放到事务中来处理，而一旦用上事务，则性能下降更加明显。

2.2 异步双写

由于发消息比写es快多了，所以异步双写通过发消息来异步执行写es的逻辑！
优点：

性能高
不存在丢失数据的问题

缺点：

还存在硬编码、业务强耦合等问题；
系统中增加了mq的代码，复杂度增加；
可能存在时延问题，程序的写入性能提高了，但是由于MQ的消费可能由于网络或其它原因导致用户写入的数据不一定可以马上看到。

2.3 定时任务调度

上面两种方案中都存在硬编码问题，也就是有任何对mysq进行增删改查的地方要么植入ES代码，要么替换为MQ代码，代码的侵入性太强，若是实时要求不高的情况下，可以考虑用定时器来处理，具体步骤如下：

数据库的相关表中增加一个字段为timestamp的字段，任何crud操作都会导致该字段的时间发生变化；
原来程序中的crud操作不做任何变化；
增加一个定时器程序，让该程序按一定的时间周期扫描指定的表，把该时间段内发生变化的数据提取出来；
逐条写入到ES中。

优点：

不改变原来代码，没有侵入性、没有硬编码；
没有业务强耦合；
不改变原来程序的性能；
任务调度代码编写简单不需要考虑增删改查。

缺点：

时效性较差，由于定时器工作周期不可能设在秒级，所以实时性没有上面2中好；
对数据库有一定的轮询压力，一种改进方法是将轮询放到压力不大的从库上。

2.4 Binlog同步

上面三种方案要不有代码侵入、要不有硬编码、要不有时延，那么有没有一种稍微折中的方法嘞？答案就是利用mysql的binlog！
具体步骤如下：

读取mysql的binlog日志，获取指定表的日志信息；
将读取的信息转为MQ；
编写一个MQ消费程序；
不断消费MQ，每消费完一条消息，将消息写入到ES中。

优点：

没有代码侵入、没有硬编码；
原有系统不需要任何变化，没有感知；
性能高
业务解耦，不需要关注原来系统的业务逻辑

缺点：

构建Binlog系统复杂
存在MQ延时的风险

3. 数据同步方案和失败处理

3.1 我的数据同步方案

我们重新看一下我们要解决的问题：高效且稳定地将mysql的多表数据同步到ES宽表。通过之前的铺垫，我们权衡利弊，选择Binlog同步的方式来解决这个问题。因为目前已经有很多数据同步工具，比如cannal和DTS，我们这才用DTS来完成数据同步任务，整体思路如下所示：

第一步：通过DTS数据订阅服务来订阅mysql（涉及的业务表）的增量日志
第二步：根据业务需求，在消费应用中实现数据定制化消费，对mysql多表数据进行加工，写入mysql的一张宽表
第三步：通过DTS数据同步服务实现mysql中的宽表数据和ES的数据同步，这一步我们是无条件信任的

加入mysql的宽表来中转而不选择消费应用直接写入ES，可以保证整个数据同步过程的稳定性。我们无条件信任DTS同步过程，但是消费应用写入宽表mysql的过程是有风险的，宽表mysql能够让我们快速定位错误。

这里的RDS你可以理解为mysql，是阿里云的云服务版本罢了！

3.2 失败处理

这里有个面试经常问的点，就是消息消费失败，你如何保证数据的一致性。或者说换个问法你有什么失败处理策略嘛！
因为我们的消费应用承担的职责是数据构造和数据存储。dts订阅版本会监听业务数据的变动，然后会发送一条消息给我们的消费应用，我们的消费应用会根据变动的主键id去查表构造一条全量数据插入大宽表。因为这个消费过程是我们人工代码去实现的，如果消费失败怎么办？我主要采用了本地消息表和定时任务来进行补偿，从而保证数据的最终一致性。
本地消息表是一种应对分布式系统中消息消费失败或者业务处理失败的解决方案，它通过在本地记录待处理消息的相关信息，然后利用分布式调度框架处理这些没有消费成功的消息，保证业务的完整性和正确性。
第一步我们要建一张消息记录表，本地消息表主要有如下字段：

id：消息唯一标识符，可以使用 UUID 或者自增长整数类型；
topic: 消息topic
tag: 消息tag
message：待发送的消息内容，可以是文本、JSON 等格式；
status：消息状态，可以是以下几种状态之一：

new：新建状态，待发送或者处理；
processing：处理中状态，消息正在处理中；
success：处理成功状态，消息处理成功；
fail：处理失败状态，消息处理失败；

retries：消息重试次数，记录消息已经尝试重新发送或者处理的次数；
next_retry_time：下一次重试时间，记录下一次重试的时间，可以使用时间戳或者日期时间格式；
create_time：消息创建时间，记录消息的创建时间，可以使用时间戳或者日期时间格式；
update_time：消息更新时间，记录消息的最后更新时间，可以使用时间戳或者日期时间格式。

第二步我们写一个基于注解的切面，我们在所有消费方法上添加切面注解@RetryMSg，这个注解在消费方法执行完之后会对根据消费成功与否做出相应逻辑：

消费成功就查一下本地消息表，如果本地消息表没有数据，说明这条消息第一次消费就成功了，那就不管；如果本地消息表有数据，说明这条消息的消费实在重试过程中成功的，那么就更改这条消息在本地消息表中的状态即可！
消息失败，先查一下本地消息表有没有该消息，如果没有就新增一条消息失败信息进去即可

第三步我们要写定时任务SyncEsRetryExecutor�了，这个定时任务每隔一段时间（自己设置）会拉去状态为new和fail的所有消息进行任务重试，将构造的数据重新插入大宽表，如果插入成功就在本地消息表修改一下状态，如果重试五次之后还没有消费成功，那么就要触发钉钉机器人告警了，相关人员会在值班群里面看到告警信息，然后人工进行数据订正。