详解 canal 同步 MySQL 增量数据到 ES( 二 ) _MySQL

2、instance 配置文件
在 conf 目录下创建实例目录 product-syn , 在 product-syn 目录创建配置文件：instance.properties 。
#按需修改成自己的数据库信息#################################################...canal.instance.master.address=192.168.1.20:3306# username/password,数据库的用户名和密码...canal.instance.dbUsername = canalcanal.instance.dbPassword = canal...# table regex canal.instance.filter.regex=mytest.t_product# mq configcanal.mq.topic=product-syn-topic# 针对库名或者表名发送动态topic#canal.mq.dynamicTopic=mytest,.*,mytest.user,mytest\..*,.*\..*canal.mq.partitinotallow=0# hash partition config#canal.mq.partitinotallow=3#库名.表名: 唯一主键，多个表之间用逗号分隔#canal.mq.partitinotallow=mytest.person:id,mytest.role:id#################################################3、服务启动
启动两个 canal 服务，我们从 zookeeper gui 中查看服务运行情况。

文章插图
修改一条 t_product 表记录，可以从 RocketMQ 控制台中观测到新的消息。

文章插图
6 消费者1、产品索引操作服务

文章插图
2、消费监听器

文章插图
消费者逻辑重点有两点：

顺序消费监听器
将消息数据转换成 JSON 字符串，从 data 节点中获取表最新数据（批量操作可能是多条）。然后根据操作类型 UPDATE、 INSERT、DELETE 执行产品索引操作服务的方法。

7 写到最后canal 是一个非常有趣的开源项目，很多公司使用 canal 构建数据传输服务( Data Transmission Service ，简称 DTS )。
推荐大家阅读这个开源项目，你可以从中学习到网络编程、多线程模型、高性能队列 Disruptor、流程模型抽象等。
这篇文章涉及到的代码已收录到下面的工程中，有兴趣的同学可以一看。