开源数据同步神器——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端一般是需要大家进行简单的开发。https://github.com/alibaba/canal/wiki/ClientAPI 有一个简单的示例,很容易理解。

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"
    BuildingId: "text"
    HouseNum: "text"
    Floors: "text"
    IdProjectInfo: "text"
    HouseDigitNum: "text"
    BuildingNum: "text"
    BuildingName: "text"
    Name: "text"
    projectid: "text"
    bIdProjectInfo: "text"
    cinitid: "text"
    pCommunityId: "text"

enablefieldmap 是否需要自动生成fieldmap,默认为false,如果需要启动的时候就生成这设置为true,并且设置 fieldmap,类似elasticsearch mapping中每个字段的类型。

esconfig bug处理

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

后续计划

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

源码

源码地址:https://github.com/itmifen/canal


原文发布于微信公众号 - IT米粉(itmifen)

原文发表时间:2019-01-10

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

扫码关注云+社区

领取腾讯云代金券