- 1.引入
- 2.传统方案介绍
- 3.监控binlog实现"同步"更新
- 4.总结
1.引入
先前介绍了ElasticSearch，以及ES配合MySQL的问题，这种方案是让ES上的数据根据MySQL的数据做对照从而形成对应的索引，再将数据通过处理和封装存放在ES当中。（可回顾：技术分析 | 浅析MySQL与ElasticSearch的组合使用）回到生产环境，我们如何保证MySQL中与ES对照的数据发生更新的时候ES也进行更新呢？就以ES为例。
2.传统方案介绍
2.1直接的"同步"更新
第一种方式十分直接，当发生对MySQL数据更新操作时，由服务器对MySQL和ES同时进行更新操作，如图：
这种方式实现起来十分“简单粗暴”，容易理解。这种实现方法显然可以解决这个问题，但绝不是最优解，原因如下：
- 首先，这种方法使得我们进行数据库的数据写入、修改、删除等操作，后面都要跟上ES的同步操作，代码书写也过于冗长，且大大加大了业务的耦合度。
- 其次，这种方法不能很好解决“同步”的问题，如果在执行对应操作的时候，发生了断电等情况，就有可能导致数据不同步的问题。
- 最后，为了保证两者的更新要么同时完成要么都不完成，需要开启事务来处理，系统的性能有所降低，同时，在高并发情况下，有可能造成服务的“雪崩”。
2.2异步的"同步"更新
针对前面的方案，可以考虑加入消息队列的中间件来优化，与第一种方法不同的是当发生对MySQL数据更新操作时，服务器会完成MySQL数据的更新，并通过MQ的队列通过设置好的交换机发送更新ES的消息给对应的接收更新消息的队列，然后完成对应ES数据更新的实现。如图：
这种方案将直接的更新方式转换为异步的更新方式，性能上显然提高了，同时降低了业务耦合度，也优化了数据“同步”的问题。但是，这种方案会出现MQ的消费者在消费时可能会因为网络等原因导致用户数据有延时。同时，从编码角度上看，每次系统要进行同步的时候都要编写MQ代码，仍然存在业务的耦合，同时系统架构的设计也因为加入新的中间件要重新考虑维护的问题。
3.监控binlog实现"同步"更新
上面两种方案中都存在硬编码问题，同时存在强的业务耦合，以至于实现MySQL数据更新后的数据同步问题的代价要么是植入ES更新代码，要么替换为MQ代码，代码的侵入性太强，且性能降低。因此可以通过监控MySQL的binlog来实现数据的同步。
3.1问题分析
binlog，该日志存在于Server层次中，是使用存储引擎都可以使用的日志模块，binlog是逻辑日志，记录的是这个语句的原始逻辑，比如“给test表id=5这一行的col1字段值加1”。binlog的日志文件是可以追加写入的。“追加写入”是指binlog日志文件写到一定大小后会切换到下一个文件进行写入，可以设置sync_binlog为1，让每次事务的binlog都持久化保存到磁盘中。binlog在ROW模式下会记录每次操作后每行记录的变化。虽然此模式下所占用的空间较大，但此模式可以保持数据的一致性。因此不管SQL是什么，引用了什么函数，他记录的是执行后的效果。
3.2使用Canal来监控binlog
Canal是阿里用Java开发的基于数据库增量的日志解析，是提供增量数据订阅&消费的中间件。目前,Canal主要支持了 MySQL 的 binlog 解析，解析后可利用 Canal Client 来处理获得的相关数据。
详细可参考：https://github.com/alibaba/canal/wiki
Canal的实现原理基于MySQL主从复制进行设计：
- Master主库将改变记录到逻辑日志(binary log)中（这些记录叫做逻辑日志事件，binary log events，可以通过show binlog events进行查看）；
- Slave从库将Master主库的binary log events拷贝到它的中继日志(relay log)；
- Slave从库读取从重做中继日志中的事件，将改变反映它自己的数据同步到数据库中。
而Canal就是将自身伪装成一个Slave从库，假装从Master主库复制数据：
- Canal模拟MySQL Slave的交互协议，伪装自己为MySQL Slave，向MySQL Master发送dump协议；
- MySQL Master收到dump请求，开始推送binary log给Slave(也就是Canal)；
- Canal解析binary log对象(原始为byte流)。
这种方案的好处是程序中没有代码侵入、没有硬编码。同时，原有系统不需要任何变化对原方案的高耦合进行了业务解耦，不需要关注原来系统的业务逻辑。
对于自建 MySQL , 需要先开启 Binlog 写入功能，配置 binlog-format 为 ROW 模式，my.cnf 中配置如下：
[mysqld]# 开启 binloglog-bin=mysql-bin # 选择 ROW 模式binlog-format=ROW # 指定开启binlog的数据库，不指定则全部数据库开启binlog-do-db=databasename# 配置 MySQL replaction 需要定义，不要和 canal 的 slaveId 重复server_id=1
创建canal账户，授权 canal 链接 MySQL 账号具有作为 MySQL slave 的权限
[mysqld]CREATE USER canal IDENTIFIED BY 'canal'; GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';-- GRANT ALL PRIVILEGES ON *.* TO 'canal'@'%' ;FLUSH PRIVILEGES;
下载并启动Canal
wget https://github.com/alibaba/canal/releases/download/canal-1.1.2/canal.deployer-1.1.2.tar.gzmkdir /tmp/canaltar zxvf canal.deployer-1.1.2.tar.gz -C /tmp/canal
修改Canal的配置文件
vi conf/example/instance.properties## mysql serverIdcanal.instance.mysql.slaveId = 1234#position info，需要改成自己的数据库信息canal.instance.master.address = 127.0.0.1:3306 canal.instance.master.journal.name = canal.instance.master.position = canal.instance.master.timestamp = #canal.instance.standby.address = #canal.instance.standby.journal.name =#canal.instance.standby.position = #canal.instance.standby.timestamp = #username/password，需要改成自己的数据库信息canal.instance.dbUsername = canal canal.instance.dbPassword = canalcanal.instance.defaultDatabaseName =canal.instance.connectionCharset = UTF-8#table regexcanal.instance.filter.regex = .*\\..*
Canal操作
# 启动sh bin/startup.sh# 查看server日志vi logs/canal/canal.log</pre># 查看 instance 的日志vi logs/example/example.log# 关闭sh bin/stop.sh
以Java为例，创建测试项目Maven工程，导入应用开发场景：
<dependencies> <dependency> <groupId>com.alibaba.otter</groupId> <artifactId>canal.client</artifactId> <version>1.1.2</version> </dependency> <dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>2.4.1</version> </dependency> </dependencies>
编写日志监视类CanalClient来从日志中抓取信息，首先，获取canal的连接对象并连接：
//获取 canal 连接对象CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1", 11111),"example", "canal","canal");//连接canalConnector.connect();
指定需要监控的数据库，并根据数据量来获取 Message ：
//指定要监控的数据库canalConnector.subscribe("databasename.*");//获取 MessageMessage message = canalConnector.get(100);
接着就可以通过处理 Message 来得到监控信息内容了：
List<CanalEntry.Entry> entries = message.getEntries(); if (entries.size() > 0) { for (CanalEntry.Entry entry : entries) { //获取表名 String tableName = entry.getHeader().getTableName(); //Entry 类型 CanalEntry.EntryType entryType = entry.getEntryType(); //判断 entryType 是否为 ROWDATA if (CanalEntry.EntryType.ROWDATA.equals(entryType)) { //序列化数据 ByteString storeValue = entry.getStoreValue(); //反序列化 CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue); //获取事件类型 CanalEntry.EventType eventType = rowChange.getEventType(); //获取具体的数据 List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList(); //遍历并打印数据 for (CanalEntry.RowData rowData : rowDatasList) { List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList(); JSONObject beforeData = new JSONObject(); for (CanalEntry.Column column : beforeColumnsList) { beforeData.put(column.getName(), column.getValue()); } JSONObject afterData = new JSONObject(); List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList(); for (CanalEntry.Column column : afterColumnsList) { afterData.put(column.getName(), column.getValue()); } System.out.println("TableName:" + tableName + ",EventType:" + eventType + ",Before:" + beforeData + ",After:" + afterData); } } }}
从代码中可以看出，当系统与Canal建立连接后可以获取Message来监控数据库的操作，Message是一次Canal从MySQL的 bin log 中抓取的信息，一个Message中可以有多个SQL执行的结果，每个SQL执行结果（SQL命令）称为Entry，如图：
Entry中包含 TableName 、 EntryType 和 StoreValue ，其中 StoreValue 包含了数据变化的内容。如下：
要想进行使用还需要进行反序列化操作才可以进行使用，如下：
当然，实际生产环境Canal可以配置MQ模式，配合RocketMQ或者Kafka，canal会把数据发送到MQ的topic中，然后通过消息队列的消费者进行处理。首先需要修改canal.properties文件,这个文件是 canal 的基本通用配置，canal 端口号默认就是 11111，修改 canal 的输出 model，默认 tcp，改为输出到 kafka，instance.properties文件输出得到主题为kafka，可配置集群，再次启动canal就可以启动 Kafka 消费客户端测试，查看消费情况了。
4.总结
本文介绍了三种方式使得中间件的数据与MySQL的数据保存同步，前两种方法在使用性能和设计上都存在较大漏洞，而第三种通过读取MySQL的bin log日志，获取指定表的日志信息来实现数据同步的方法，在编码上看没有代码侵入，业务耦合度低，且原有系统不需要任何变化，但，构建bin log监控系统需要做好规划，不多赘述了。
Enjoy GreatSQL 🙂
