开源数据同步神器——canal

前言

现在大型的IT系统中，都会使用分布式的方式，同时会有很是多的中间件，如redis、消息队列、大数据存储等，可是实际核心的数据存储依然是存储在数据库，做为使用最普遍的数据库，如何将mysql的数据与中间件的数据进行同步，既能确保数据的一致性、及时性，也能作到代码无侵入的方式呢？若是有这样的一个需求，数据修改后，须要及时的将mysql中的数据更新到elasticsearch,咱们会怎么进行实现呢？

数据同步方案选择

针对上文的需求，通过思考，初步有以下的一些方案：

• 代码实现
  针对代码中进行数据库的增删改操做时，同时进行elasticsearch的增删改操做。

• mybatis实现
  经过mybatis plugin进行实现，截取sql语句进行分析， 针对insert、update、delete的语句进行处理。显然，这些操做若是都是单条数据的操做，是很容易处理的。可是，实际开发中，老是会有一些批量的更新或者删除操做，这时候，就很难进行处理了。

• Aop实现
  不论是经过哪一种Aop方式，根据制定的规则，如规范方法名，注解等进行切面处理，但依然仍是会出现没法处理批量操做数据的问题。

• logstash
  logstash相似的同步组件提供的文件和数据同步的功能，能够进行数据的同步，只须要简单的配置就能将mysql数据同步到elasticsearch，可是logstash的原理是每秒进行一次增量数据查询，将结果同步到elasticsearch，实时性要求特别高的，可能没法知足要求。且此方案的性能不是很好，形成资源的浪费。

实现方式	优缺点
代码实现	技术难度低，侵入性强，实时性高
基于mybatis	有必定的技术难度，可是没法覆盖全部的场景
Aop实现	技术难度低，半侵入性，须要规范代码，依然没法覆盖全部的场景
logstash	技术难度低，无侵入性，无需开发，但会形成资源浪费。

那么是否有什么更好的方式进行处理吗？mysql binlog同步，实时性强，对于应用无任何侵入性，且性能更好，不会形成资源浪费，那么就有了我今天的主角——canal

canal

介绍

canal 是阿里巴巴的一个开源项目，基于java实现，总体已经在不少大型的互联网项目生产环境中使用，包括阿里、美团等都有普遍的应用，是一个很是成熟的数据库同步方案，基础的使用只须要进行简单的配置便可。
canal是经过模拟成为mysql 的slave的方式，监听mysql 的binlog日志来获取数据，binlog设置为row模式之后，不只能获取到执行的每个增删改的脚本，同时还能获取到修改前和修改后的数据，基于这个特性，canal就能高性能的获取到mysql数据数据的变动。

使用

canal的介绍在官网有很是详细的说明，若是想了解更多，你们能够移步官网（https://github.com/alibaba/canal）了解。我这里补充下使用中不太容易理解部分。
canal的部署主要分为server端和client端。
server端部署好之后，能够直接监听mysql binlog,由于server端是把本身模拟成了mysql slave，因此，只能接受数据，没有进行任何逻辑的处理，具体的逻辑处理，须要client端进行处理。
client端通常是须要你们进行简单的开发。github.com/alibaba/can… 有一个简单的示例，很容易理解。

canal Adapter

为了便于你们的使用，官方作了一个独立的组件Adapter，Adapter是能够将canal server端获取的数据转换成几个经常使用的中间件数据源，如今支持kafka、rocketmq、hbase、elasticsearch，针对这几个中间件的支持，直接配置便可，无需开发。上文中，若是须要将mysql的数据同步到elasticsearch，直接运行 canal Adapter，修改相关的配置便可。

常见问题

• 没法接收到数据，程序也没有报错？
  必定要确保mysql的binlog模式为row模式，canal原理是解析Binlog文件，而且直接中文件中获取数据的。

• Adapter 使用没法同步数据？
  按照官方文档，检查配置项，如sql的大小写，字段的大小写可能都会有影响，若是还没法搞定，能够本身获取代码调试下，Adapter的代码仍是比较容易看懂的。

canal Adapter elasticsearch 改造

由于有了canal和canal Adapter这个神器，同步到elasticsearch、hbase等问题都解决了，可是本身的开发的过程当中发现，Adapter使用仍是有些问题，由于先使用的是elasticsearch同步功能，因此对elasticsearch进行了一些改造：

elasticsearch初始化

一个全新的elasticsearch没法使用，由于没有建立elasticsearch index和mapping,增长了对应的功能。 elasticsearch配置文件mapping节点增长两个参数：

 enablefieldmap: true
 fieldmap:
 id: "text"
 name: "text"
 c_time: "text"
    
复制代码

enablefieldmap 是否须要自动生成fieldmap，默认为false,若是须要启动的时候就生成这设置为true,而且设置 fieldmap,相似elasticsearch mapping中每一个字段的类型。

esconfig bug处理

代码中获取binlog的日志处理时，必需要获取数据库名，可是当获取binlog为type query时，是没法获取 数据库名的，此处有bug，致使出现 "Outer adapter write failed" ,且未输出错误日志，修复此bug.

后续计划

• 增长rabbit MQ的支持
• 增长redis的支持

源码

源码地址：github.com/itmifen/can…

---

微信号：itmifen