先说下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:\ProgramData\MySQL\MySQL Server 5.6的文件夹下(如果没有,可能是没有将隐藏的文件夹显示),而不是这个路径C:\Program Files\MySQL\MySQL 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=FILE12 general-log=1 (只需要将0更改为1即可)13 general_log_file="MYUNYU.log"14 slow-query-log=115 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:\...\canal\canal.deployer-1.1.1\conf\example的文件夹,打开配置文件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:330610 canal.instance.master.journal.name=11 canal.instance.master.position=12 canal.instance.master.timestamp=13 canal.instance.master.gtid=14 15 # rds oss binlog16 canal.instance.rds.accesskey=17 canal.instance.rds.secretkey=18 canal.instance.rds.instanceId=19 20 # table meta tsdb info21 canal.instance.tsdb.enable=true22 #canal.instance.tsdb.url=jdbc:mysql://127.0.0.1:3306/springboot_test123 #canal.instance.tsdb.dbUsername=canal24 #canal.instance.tsdb.dbPassword=canal25 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/password33 canal.instance.dbUsername=canal34 canal.instance.dbPassword=canal35 canal.instance.connectionCharset = UTF-8 #这里配置监控的数据库名36 canal.instance.defaultDatabaseName =springboot_test137 # enable druid Decrypt database password38 canal.instance.enableDruid=false39 #canal.instance.pwdPublicKey=MFwwDQYJKoZIhvcNAQEBBQADSwAwSAJBALK4BUxdDltRRE5/zXpVEVPUgunvscYFtEip3pmLlhrWpacX7y7GCMo2/JM6LeHmiiNdH1FWgGCpUfircSwlWKUCAwEAAQ==40 41 # table regex42 canal.instance.filter.regex=.*\\..*43 # table black regex44 canal.instance.filter.black.regex=45 #################################################
再进入路径为C:\...\canal\canal.deployer-1.1.1\bin的文件夹,双击打开startup.bat,如果显示有以下内容,则表示配置成功,canal服务器端启动ok:
4、客户端代码测试:
(1)首先创建一个空的maven项目,在pom文件中引入canal客户端的依赖:
12 com.alibaba.otter 3canal.client 41.0.12 5
(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 // 创建链接 instanceA20 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( Listentrys) {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 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的数据写入进行监控的操作实现完毕!