关系型数据库同步数据到Elasticsearch设计方案

1、构建canal服务订阅数据库binlog消息canal服务监听到数据库表的变化，发送消息给同步MQ

2、同步MQ消息消费同步MQ在接受canal发出的binlog消息之后，对binlog消息进行解析，获取到需要写入ES索引的主键，根据主键反查数赍铈于脏据库，组织写入ES的数据，通过ES客户端创建ES索引文档反查数据库的原因是MQ消息的无序性，保证ES中保存的信息是DB中最新的信息，防止出现ABA问题

3、写入es成功，发送消蝌怦吩曰息给对账程序这里我们采用延时对账的方式，保证DB中数据和ES中数据的一致性在写入es成功之后，发送延时消息（谮燥寻寒延时2s或者5s，根据测试结果而定）给对账程序，对账程序有2种对账方式，第一种是全量对账，意思就是以DB为主对比ES中的每个字段是否和DB中一致，第二种是对比表更新时间，意思就是对比DB表中该条数据的更新时间是否和ES中保存的更新时间一致。全量对账适合ES聚合了多个表的数据，且表中数据更新频繁；对比更新时间的方式适合ES索引结构简单但是字段较多对账不一致把该条MQ消息发给异常MQ队列，重新同步该条消息到ES

4、写入es失败，或者写入程序有异常，则把该条MQ消息发送给异常MQ队列之所以使用新创建一个MQ队列来处理异常消息，而不是同步MQ队列消息重试，是为了尽可能的减少同步MQ的消息量，防止消息出现积压，同时在手动刷新ES数据的时候，也可以使用异常MQ队列

5、手动刷新ES接口有的人会说，已经监听了binlog消息，为什么还需要手动刷新ES接口，是因为在很多时候我们需要不想要需要数据库中的任何一个字段制造消息变更，还有不想要因为刷新ES的数据而造成同步MQ队列积压，影响线上正常业务