技术|基于canal实现mysql和es增量数据同步 mysql|java|elasticsearch

一、简介

canal主要用途是基于 MySQL 数据库增量日志解析，提供增量数据订阅和消费。

二、工作原理

canal 模拟 MySQL slave 的交互协议，伪装自己为 MySQL slave ，向 MySQL master 发送 dump 协议MySQL master 收到 dump 请求，开始推送 binary log 给 slave (即 canal )canal 解析 binary log 对象(原始为 byte 流）

三、基于canal实现mysql和es增量数据同步
3.1 基于canalServer和canalAdapter的方式进行数据同步
worker：数据全量同步组件
canalServer：接收binlog信息
canalAdapter：canalServer和elasticSearch适配
canal作为增量数据同步的工具，同时对不同数据源做了适配，canalAdapter目前默认支持的数据源有：Hbase，ElasticSearch和RDB（关系型数据库）。
通过canalServer和canalAdapter的一些配置（各个中间件配置信息请参考官方文档），无需代码开发就能实现不同数据源之间的同步。
但是用这种方式如果数据量大，可能会产生数据积压，有数据丢失的风险。
3.2基于mq和canalClient的方式进行数据同步
canal 1.1.1版本之后, 默认支持将canalServer接收到的binlog数据直接投递到MQ, 目前默认支持的MQ系统有:kafka和RocketMQ。
我们需要在应用中开发canalClient，esClient和mqConsumer组件
canalClient：将消息解析并封装成对应实体
esClient：elasticSearch基本操作
mqConsumer：消费消息，通过canalClient将消息解析成对应的实体对象，通过esClient将对象写入到elasticSearch集群。
canalClient的关键代码解析Message中的Entries，代码如下：

private static void printEntry(List entrys) { for (Entry entry : entrys) { if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) { continue; } RowChange rowChage; try { rowChage = RowChange.parseFrom(entry.getStoreValue()); } catch (Exception e) { throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e); } //获取操作类型：insert/update/delete类型 EventType eventType = rowChage.getEventType(); //打印Header信息 System.out.println(String.format("================》; binlog[%s:%s] , name[%s,%s] , eventType : %s", entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(), entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType)); //判断是否是DDL语句 if (rowChage.getIsDdl()) { System.out.println("================》; isDdl: true,sql:" + rowChage.getSql()); } //获取RowChange对象里的每一行数据，打印出来 for (RowData rowData : rowChage.getRowDatasList()) { //如果是删除语句 if (eventType == EventType.DELETE) { printColumn(rowData.getBeforeColumnsList()); //如果是新增语句 } else if (eventType == EventType.INSERT) { printColumn(rowData.getAfterColumnsList()); //如果是更新的语句 } else { //变更前的数据 printColumn(rowData.getBeforeColumnsList()); //变更后的数据 printColumn(rowData.getAfterColumnsList()); } } } }

四、小结
canal只适合做增量数据同步，全量数据同步需要采用其他的方式来实现。
【技术|基于canal实现mysql和es增量数据同步】?