博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
关于使用Binlog和canal来对MySQL的数据写入进行监控
阅读量:6126 次
发布时间:2019-06-21

本文共 8845 字,大约阅读时间需要 29 分钟。

先说下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客户端的依赖:

1     
2
com.alibaba.otter
3
canal.client
4
1.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( List
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
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的数据写入进行监控的操作实现完毕!

 

转载于:https://www.cnblogs.com/mYunYu/p/9935225.html

你可能感兴趣的文章
python实现牛顿法求解求解最小值(包括拟牛顿法)【最优化课程笔记】
查看>>
js中var、let、const的区别
查看>>
腾讯云加入LoRa联盟成为发起成员,加速推动物联网到智联网的进化
查看>>
从Python2到Python3:超百万行代码迁移实践
查看>>
Windows Server已可安装Docker,Azure开始支持Mesosphere
查看>>
简洁优雅地实现夜间模式
查看>>
react学习总结
查看>>
微软正式发布PowerShell Core 6.0
查看>>
Amazon发布新的会话管理器
查看>>
InfoQ趋势报告:DevOps 和云计算
查看>>
舍弃Python,为什么知乎选用Go重构推荐系统?
查看>>
在soapui上踩过的坑
查看>>
MySQL的字符集和字符编码笔记
查看>>
ntpd同步时间
查看>>
must implement java.io.Serializable hessian
查看>>
Microsoft Licenses Flash Lite for Windows Mobile Users
查看>>
HDOJ 2020 绝对值排序
查看>>
HDOJ/HDU 2560 Buildings(嗯~水题)
查看>>
Maven编译时跳过Test
查看>>
Spring Boot 整合Spring Security 和Swagger2 遇到的问题小结
查看>>