2 监听mysql表内容变化,使用canal

您所在的位置:网站首页 监听数据库字段的变化 2 监听mysql表内容变化,使用canal

2 监听mysql表内容变化,使用canal

2023-11-14 02:16| 来源: 网络整理| 查看: 265

mysql本身是支持主从的(master slave),原理就是master产生的binlog日志记录了所有的增删改语句,将binlog发送到slave节点进行执行即可完成数据的同步。

canal是阿里开源的一个中间件,它就是通过解析binlog来完成数据变更的监听的。

https://github.com/alibaba/canal

可以看到,canal是这样工作的:canal有一个server工程,该server自己伪装为mysql的一个slave节点,然后向master请求所有的变更日志。同时canal有一个client工程,通过添加client的sdk,我们就可以在项目里监听到server端传来的数据变更信息,从而达到监听数据变化的目的。

安装配置canal

在这里https://github.com/alibaba/canal/releases,下载canal releases包,如https://github.com/alibaba/canal/releases/download/canal-1.0.22/canal.deployer-1.0.22.tar.gz。

这个就是server端,下载后解压,如图

打开conf/example下的instance.properties文件,编辑

################################################# ## mysql serverId canal.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 = canal canal.instance.defaultDatabaseName = test canal.instance.connectionCharset = UTF-8 # table regex canal.instance.filter.regex = .*\\..* # table black regex canal.instance.filter.black.regex = #################################################

address设置为mysql的连接地址,defaultDatabaseName设置为自己要监听的库名,如test。

在mysql命令行,创建一个新用户,作为slave

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

对应配置文件里的canal用户。到此配置完毕。

执行bin目录下的startup.sh

启动后可以在logs目录下查看日志。在example目录下的example.log,如果没有报错,说明启动成功。

canal客户端编写

服务端启动完毕后,在客户端即可监听test库的变化。

新建一个java maven项目,pom.xml里添加依赖

com.alibaba.otter canal.client 1.0.12 package com.mindata; import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnectors; import com.alibaba.otter.canal.common.utils.AddressUtils; import com.alibaba.otter.canal.protocol.CanalEntry; import com.alibaba.otter.canal.protocol.Message; import java.net.InetSocketAddress; import java.util.List; /** * A Camel Application */ public class MainApp { /** * A main() so we can easily run these routing rules in our IDE */ public static void main(String... args) throws Exception { // 创建链接 CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(AddressUtils.getHostIp(), 11111), "example", "", ""); int batchSize = 1000; int emptyCount = 0; try { connector.connect(); connector.subscribe(".*\\..*"); connector.rollback(); int totalEmptyCount = 120; while (emptyCount < totalEmptyCount) { Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据 long batchId = message.getId(); int size = message.getEntries().size(); if (batchId == -1 || size == 0) { emptyCount++; System.out.println("empty count : " + emptyCount); try { Thread.sleep(1000); } catch (InterruptedException e) { } } else { emptyCount = 0; // System.out.printf("message[batchId=%s,size=%s] \n", batchId, size); printEntry(message.getEntries()); } connector.ack(batchId); // 提交确认 // connector.rollback(batchId); // 处理失败, 回滚数据 } System.out.println("empty too many times, exit"); } finally { connector.disconnect(); } } private static void printEntry(List entrys) { for (CanalEntry.Entry entry : entrys) { if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry .EntryType .TRANSACTIONEND) { continue; } CanalEntry.RowChange rowChage = null; try { rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue()); } catch (Exception e) { throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e); } CanalEntry.EventType eventType = rowChage.getEventType(); System.out.println(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s", entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(), entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType)); for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) { if (eventType == CanalEntry.EventType.DELETE) { printColumn(rowData.getBeforeColumnsList()); } else if (eventType == CanalEntry.EventType.INSERT) { printColumn(rowData.getAfterColumnsList()); } else { System.out.println("-------> before"); printColumn(rowData.getBeforeColumnsList()); System.out.println("-------> after"); printColumn(rowData.getAfterColumnsList()); } } } } private static void printColumn(List columns) { for (CanalEntry.Column column : columns) { System.out.println(column.getName() + " : " + column.getValue() + " update=" + column.getUpdated()); } } }

注意newSingleConnector那里,设置canal server的ip和端口,端口默认为11111。example是和conf目录下的相对应。

如果你在这里新建个目录abc,目录里也放一个instance.properties,那么就可以在客户端监听abc。

启动客户端,查看控制台输出。

启动后,就可以打印empty count,此时你可以操作数据库里的test库,做任何增删改的操作,然后就能在控制台看到打印的语句。

可以看到,修改的是test库里的History表,before代表修改前,id=3,calltime字段值为3,after修改后,calltime为22.customer列我没有修改,所以值为242没变。

这样就获得了对应的库里,某个表的任何一列的变化的解析。



【本文地址】


今日新闻


推荐新闻


CopyRight 2018-2019 办公设备维修网 版权所有 豫ICP备15022753号-3