关于使用Binlog和canal来对MySQL的数据写入进行监控

先说下Binlog和canal是什么吧。

1、Binlog是mysql数据库的操作日志，当有发生增删改查操作时，就会在data目录下生成一个log文件，形如mysql-bin.000001，mysql-bin.000002等格式

2、canal是阿里巴巴旗下的一款开源项目，纯Java开发。基于数据库增量日志解析，提供增量数据订阅&消费，目前主要支持了MySQL（也支持mariaDB）；

3、canal起源：早期，阿里巴巴B2B公司因为存在杭州和美国双机房部署，存在跨机房同步的业务需求。不过早期的数据库同步业务，主要是基于trigger的方式获取增量变更，不过从2010年开始，阿里系公司开始逐步的尝试基于数据库的日志解析，获取增量变更进行同步，由此衍生出了增量订阅&消费的业务，从此开启了一段新纪元。

4、基于日志增量订阅&消费支持的业务：

（1）数据库镜像

（2）数据库实时备份

（3）多级索引 (卖家和买家各自分库索引)

（4）search build

（5）业务cache刷新

（6）价格变化等重要业务消息

二、工作原理：

1、mysql主备复制实现：

从上层来看，复制分成三步：

（1）master将改变记录到二进制日志(binary log)中（这些记录叫做二进制日志事件，binary log events，可以通过show binlog events进行查看）；

（2）slave将master的binary log events拷贝到它的中继日志(relay log)；

（3）slave重做中继日志中的事件，将改变反映它自己的数据。

2、canal的工作原理：

原理相对比较简单：

（1）canal模拟mysql slave的交互协议，伪装自己为mysql slave，向mysql master发送dump协议

（2）mysql master收到dump请求，开始推送binary log给slave(也就是canal)

（3）canal解析binary log对象(原始为byte流)

三、主要配置：

1、mysql默认是没有开启Binlog，不妨查看下本地mysql是否开启，可执行：

1 SHOW VARIABLES LIKE 'log_%';

如下图，是我本地mysql，“on”代表已经开启，“off”代表关闭

2、如何开启Binlog：

（1）先进入路径为C:ProgramDataMySQLMySQL Server 5.6的文件夹下(如果没有，可能是没有将隐藏的文件夹显示)，而不是这个路径C:Program FilesMySQLMySQL Server 5.6

找到my.ini文件，在打开之前需要先将mysql服务停止，之后在my.ini配置文件中添加以下内容：

 1 #添加这一行就ok  
 2 log-bin=mysql-bin 
 3 #选择row模式 
 4 binlog-format=ROW  
 5 #配置mysql replaction需要定义，不能和canal的slaveId重复  
 6 server_id=1 
 7 
 8 #指定生成log的数据库
 9 binlog_do_db=springboot_test1（这是指定需要生成log的数据库，如果删除这句，则表示所有数据库都需要生成log）
10 
11 log-output=FILE
12 general-log=1  （只需要将0更改为1即可）
13 general_log_file="MYUNYU.log"
14 slow-query-log=1
15 slow_query_log_file="MYUNYU-slow.log"
16 long_query_time=10

配置好之后，再启动mysql服务，执行查看binlog是否开启，如果还没开启，那就可能是配置出了问题或者mysql版本的问题，这里不详细说

（2）从节点通过一个专门的账号连接主节点，这个账号需要拥有全局的 REPLICATION 权限。我们可以使用GRANT 命令创建这样的账号（需要先选中mysql系统库）：

1 CREATE USER canal IDENTIFIED BY 'canal';  
2 GRANT SELECT, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'canal'@'%';
3 FLUSH PRIVILEGES;

再查询mysql系统库中的user表是否存在canal用户：

3、配置canal：

首先先下载canal服务端代码canal.deployer-1.1.1.tar.gz（https://github.com/alibaba/canal/releases），解压之后，配置文件在conf文件夹下，

进入路径为C:...canalcanal.deployer-1.1.1confexample的文件夹，打开配置文件instance.properties,详细配置如下：

 1 #################################################
 2 ## mysql serverId , v1.0.26+ will autoGen 
　　#slaveId不能与my.ini配置文件的server_id一致
 3 canal.instance.mysql.slaveId=1234
 4 
 5 # enable gtid use true/false
 6 canal.instance.gtidon=false
 7 
 8 # position info
 9 canal.instance.master.address=127.0.0.1:3306
10 canal.instance.master.journal.name=
11 canal.instance.master.position=
12 canal.instance.master.timestamp=
13 canal.instance.master.gtid=
14 
15 # rds oss binlog
16 canal.instance.rds.accesskey=
17 canal.instance.rds.secretkey=
18 canal.instance.rds.instanceId=
19 
20 # table meta tsdb info
21 canal.instance.tsdb.enable=true
22 #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/springboot_test1
23 #canal.instance.tsdb.dbUsername=canal
24 #canal.instance.tsdb.dbPassword=canal
25 
26 #canal.instance.standby.address =
27 #canal.instance.standby.journal.name =
28 #canal.instance.standby.position =
29 #canal.instance.standby.timestamp =
30 #canal.instance.standby.gtid=
31 
32 # username/password
33 canal.instance.dbUsername=canal
34 canal.instance.dbPassword=canal
35 canal.instance.connectionCharset = UTF-8
　　#这里配置监控的数据库名
36 canal.instance.defaultDatabaseName =springboot_test1
37 # enable druid Decrypt database password
38 canal.instance.enableDruid=false
39 #canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==
40 
41 # table regex
42 canal.instance.filter.regex=.*\..*
43 # table black regex
44 canal.instance.filter.black.regex=
45 #################################################

再进入路径为C:...canalcanal.deployer-1.1.1in的文件夹，双击打开startup.bat，如果显示有以下内容，则表示配置成功，canal服务器端启动ok：

4、客户端代码测试：

（1）首先创建一个空的maven项目，在pom文件中引入canal客户端的依赖：

1 　　　　<dependency>
2             <groupId>com.alibaba.otter</groupId>
3             <artifactId>canal.client</artifactId>
4             <version>1.0.12</version>
5         </dependency>

（2）创建一个类进行测试：

 1 package com.xxx;
 2 
 3 import java.net.InetSocketAddress;
 4 import java.util.List;
 5 
 6 import com.alibaba.otter.canal.client.CanalConnector;
 7 import com.alibaba.otter.canal.protocol.Message;
 8 import com.alibaba.otter.canal.protocol.CanalEntry.Column;
 9 import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
10 import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
11 import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
12 import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
13 import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
14 import com.alibaba.otter.canal.client.*;
15 
16 public class CanalClient {
17 
18     public static void main(String args[]) {
19         // 创建链接            instanceA
20         CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("127.0.0.1",
21                 11111), "example", "", "");
22         int batchSize = 1000;
23         int emptyCount = 0;
24         try {
25             connector.connect();
26             connector.subscribe(".*\..*");
27             connector.rollback();
28             int totalEntryCount = 1200;
29             while (emptyCount < totalEntryCount) {
30                 Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
31                 long batchId = message.getId();
32                 int size = message.getEntries().size();
33                 if (batchId == -1 || size == 0) {
34                     emptyCount++;
35                     System.out.println("empty count : " + emptyCount);
36                     try {
37                         Thread.sleep(1000);
38                     } catch (InterruptedException e) {
39                         e.printStackTrace();
40                     }
41                 } else {
42                     emptyCount = 0;
43                     printEntry(message.getEntries());
44                 }
45                 connector.ack(batchId); // 提交确认
46             }
47             System.out.println("empty too many times, exit");
48         }catch (Exception e){
49             //connector.rollback(batchId); // 处理失败, 回滚数据
50         }
51         finally {
52             connector.disconnect();
53         }
54     }
55 
56     private static void printEntry( List<Entry> entrys) {
57         for (Entry entry : entrys) {
58             if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
59                 continue;
60             }
61             RowChange rowChage = null;
62             try {
63                 rowChage = RowChange.parseFrom(entry.getStoreValue());
64             } catch (Exception e) {
65                 throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
66             }
67 
68             EventType eventType = rowChage.getEventType();
69             System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
70                     entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
71                     entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
72                     eventType));
73             for (RowData rowData : rowChage.getRowDatasList()) {
74                 if (eventType == EventType.DELETE) {
75                     printColumn(rowData.getBeforeColumnsList());
76                 } else if (eventType == EventType.INSERT) {
77                     printColumn(rowData.getAfterColumnsList());
78 
79                 } else {
80                     System.out.println("-------> before");
81                     printColumn(rowData.getBeforeColumnsList());
82                     System.out.println("-------> after");
83                     printColumn(rowData.getAfterColumnsList());
84                 }
85             }
86         }
87     }
88 
89     private static void printColumn( List<Column> columns) {
90         for (Column column : columns) {
91             System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
92         }
93     }
94 
95 }

运行CanalClient的main方法，如果看到控制台出现以下内容则代表连接成功：

再到mysql中创建数据库springboot_test,springboot_test1,springboot_test2,再在这三个库中分别创建student表，sql语句为：

CREATE TABLE `student` (
`ID` int(11) NOT NULL AUTO_INCREMENT,
`NAME` varchar(20) NOT NULL,
`CLASS_NAME` varchar(30) NOT NULL,
`CREATE_DATE` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
`UPDATE_DATE` timestamp NOT NULL DEFAULT '0000-00-00 00:00:00' ON UPDATE CURRENT_TIMESTAMP,
PRIMARY KEY (`ID`) USING BTREE
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8 ROW_FORMAT=DYNAMIC;

执行插入语句：

1 INSERT INTO student(NAME , class_name )VALUES('student1' , 'class1');

sql执行完之后再看idea中的控制台，如果出现下面内容打印则表示可以监控数据库的写入等操作

到此，使用Binlog和canal来对MySQL的数据写入进行监控的操作实现完毕！