Java代码中,如何监控Mysql的binlog?
Python涨薪研究所
共 7715字,需浏览 16分钟
· 2021-07-20
源 / 文/ Dr Hydra
binlog
的主从复制了,但是放在我们的场景中,还有几个问题:第一,并不是需要复制所有表的数据,复制对象只有少量的几张表 第二,也是比较麻烦的,两个业务系统数据库表结构可能不一致。例如,要同步数据库1的A表中的某些字段到数据库2的B表中,在这一过程中,A表和B表的字段并不是完全相同
binlog
,因此我们就需要在代码中对binlog
进行一下监控。mysql-binlog-connector-java
,用来监控binlog
变化并获取数据,获取数据后再手动插入到另一个库的表中,基于它来实现了数据的同步。这个工具的git项目地址如下:https://github.com/shyiko/mysql-binlog-connector-java
mysql
的binlog
,binlog
是一个二进制文件,它保存在磁盘中,是用来记录数据库表结构变更、表数据修改的二进制日志。其实除了数据复制外,它还可以实现数据恢复、增量备份等功能。mysql
服务已经启用了binlog
:show variables like 'log_bin';
OFF
,表示没有启用,那么需要首先启用binlog
,修改配置文件:log_bin=mysql-bin
binlog-format=ROW
server-id=1
在配置文件中加入了 log_bin
配置项后,表示启用了binlog
binlog-format
是binlog
的日志格式,支持三种类型,分别是STATEMENT
、ROW
、MIXED
,我们在这里使用ROW
模式server-id
用于标识一个sql语句是从哪一个server
写入的,这里一定要进行设置,否则我们在后面的代码中会无法正常监听到事件
mysql
服务。再次查看是否启用binlog
,返回为ON
,表示已经开启成功。maven
坐标:<dependency>
<groupId>com.github.shyiko</groupId>
<artifactId>mysql-binlog-connector-java</artifactId>
<version>0.21.0</version>
</dependency>
public static void main(String[] args) {
BinaryLogClient client = new BinaryLogClient("127.0.0.1", 3306, "hydra", "123456");
client.setServerId(2);
client.registerEventListener(event -> {
EventData data = event.getData();
if (data instanceof TableMapEventData) {
System.out.println("Table:");
TableMapEventData tableMapEventData = (TableMapEventData) data;
System.out.println(tableMapEventData.getTableId()+": ["+tableMapEventData.getDatabase() + "-" + tableMapEventData.getTable()+"]");
}
if (data instanceof UpdateRowsEventData) {
System.out.println("Update:");
System.out.println(data.toString());
} else if (data instanceof WriteRowsEventData) {
System.out.println("Insert:");
System.out.println(data.toString());
} else if (data instanceof DeleteRowsEventData) {
System.out.println("Delete:");
System.out.println(data.toString());
}
});
try {
client.connect();
} catch (IOException e) {
e.printStackTrace();
}
}
BinaryLogClient
客户端对象,初始化时需要传入mysql
的连接信息,创建完成后,给客户端注册一个监听器,来实现它对binlog
的监听和解析。在监听器中,我们暂时只对4种类型的事件数据进行了处理,除了WriteRowsEventData
、DeleteRowsEventData
、UpdateRowsEventData
对应增删改操作类型的事件数据外,还有一个TableMapEventData
类型的数据,包含了表的对应关系,在后面的例子中再具体说明。DML
语句和DDL
语句,所以我们只需要处理我们关心的事件数据就行,否则会收到大量的冗余数据。com.github.shyiko.mysql.binlog.BinaryLogClient openChannelToBinaryLogStream
信息: Connected to 127.0.0.1:3306 at mysql-bin.000002/1046 (sid:2, cid:10)
binlog
成功,接下来,我们在数据库中插入一条数据,这里操作的数据库名字是tenant
,表是dept
:insert into dept VALUES(8,"人力","","1");
Table:
108: [tenant-dept]
Insert:
WriteRowsEventData{tableId=108, includedColumns={0, 1, 2, 3}, rows=[
[8, 人力, , 1]
]}
TableMapEventData
,通过它可以获取操作的数据库名称、表名称以及表的id
。之所以我们要监听这个事件,是因为之后监听的实际操作中返回数据中包含了表的id
,而没有表名等信息,所以如果我们想知道具体的操作是在哪一张表的话,就要先维护一个id
与表的对应关系。WriteRowsEventData
,其中记录了insert
语句作用的表,插入涉及到的列,以及实际插入的数据。另外,如果我们只需要对特定的一张或几张表进行处理的话,也可以提前设置表的名单,在这里根据表id
到表名的映射关系,实现数据的过滤,update
语句:update dept set tenant_id=3 where id=8 or id=9
Table:
108: [tenant-dept]
Update:
UpdateRowsEventData{tableId=108, includedColumnsBeforeUpdate={0, 1, 2, 3}, includedColumns={0, 1, 2, 3}, rows=[
{before=[8, 人力, , 1], after=[8, 人力, , 3]},
{before=[9, 人力, , 1], after=[9, 人力, , 3]}
]}
update
语句时,可能会作用于多条数据,因此在实际修改的数据中,可能包含多行记录,这一点体现在上面的rows
中,包含了id
为8和9的两条数据。delete
语句:delete from dept where tenant_id=3
rows
中同样返回了生效的两条数据:Table:
108: [tenant-dept]
Delete:
DeleteRowsEventData{tableId=108, includedColumns={0, 1, 2, 3}, rows=[
[8, 人力, , 3],
[9, 人力, , 3]
]}
update
操作为例,我们要对提取的数据后进行一下处理,更改上面例子中的方法:if (data instanceof UpdateRowsEventData) {
System.out.println("Update:");
UpdateRowsEventData updateRowsEventData = (UpdateRowsEventData) data;
for (Map.Entry<Serializable[], Serializable[]> row : updateRowsEventData.getRows()) {
List<Serializable> entries = Arrays.asList(row.getValue());
System.out.println(entries);
JSONObject dataObject = getDataObject(entries);
System.out.println(dataObject);
}
}
data
强制转换为UpdateRowsEventData
后,可以使用getRows
方法获取到更新的行数据,并且能够取到每一列的值。getDataObject
方法,用它来实现数据到列的绑定过程:private static JSONObject getDataObject(List message) {
JSONObject resultObject = new JSONObject();
String format = "{\"id\":\"0\",\"dept_name\":\"1\",\"comment\":\"2\",\"tenant_id\":\"3\"}";
JSONObject json = JSON.parseObject(format);
for (String key : json.keySet()) {
resultObject.put(key, message.get(json.getInteger(key)));
}
return resultObject;
}
format
字符串中,提前维护了一个数据库表的字段顺序的字符串,标识了每个字段位于顺序中的第几个位置。通过上面这个函数,能够实现数据到列的填装过程,我们再执行一条update
语句来查看一下结果:update dept set tenant_id=3,comment="1" where id=8
Table:
108: [tenant-dept]
Update:
[8, 人力, 1, 3]
{"tenant_id":3,"dept_name":"人力","comment":"1","id":8}
好文推荐
公司起诉技术总监索赔 90 万元,称其拖延研发进度
网传京东某程序员因压力太大,在商品页面置入骂人代码!京东辟谣:不关我们的事,外部商家干的!
某大厂程序员炫耀:来新加坡后,每天最多工作五六个小时,家庭年收入150万人民币,已躺平!
一键三连「分享」、「点赞」和「在看」
技术干货与你天天见~
评论
真高!比亚迪员工爆料比亚迪在越南的薪资水平:基本工资480万,全勤奖35万,交通补助20万,餐补110万,每周6天,每天10小时
上一篇:某大公司为逼迫员工离职,竟然把他的工位安排到厕所旁,没想到他直接开始记录领导的如厕时间,还发到公司大群...对此,你怎么看?--完--PS:欢迎在留言区留下你的观点,一起讨论提高。如果今天的文章让你有新的启发,欢迎转发分享给更多人。全文完,感谢你的耐心阅读。如果你还想看到我的文章,请一定给本
开发者全社区
0
太敢穿了!透视纱裙!性感火辣的身材
绝了呀今天的厂花:吴宣仪1995年1月26日,吴宣仪出生于海南省海口市,中国内地流行乐女歌手、影视演员。2016年2月,吴宣仪随宇宙少女发行首张迷你专辑正式出道。2018年4月,她参加《创造101》综艺选秀,获得第二名,成功加入火箭少女101组合。吴宣仪的颜值一直备受称赞,她的五官立体精致,皮肤白皙
逆锋起笔
0
某大公司为逼迫员工离职,竟然把他的工位安排到厕所旁,没想到他直接开始记录领导的如厕时间,还发到公司大群...
上一篇:字节的跳动职级与薪资(2024年)我们与公司间的合作,宛如两艘船只在茫茫大海上相互依靠,共同抵御风浪,携手驶向成功的彼岸。然而,当航向开始产生分歧,或是波涛汹涌的风浪改变了我们的初衷,我们或许应当冷静地选择和平分手,而非在风雨中硬撑。最近,一位网友的遭遇引起了广大职场人的关注和热议。这位网友
开发者全社区
0
金融研究 | 使用Python测量关键审计事项的「信息含量」
Tips: 公众号推送后内容只能更改一次,且只能改20字符。如果内容出问题,或者想更新内容, 只能重复推送。为了更好的阅读体验,建议阅读本文博客版, 链接地址https://textdata.cn/blog/2023-01-13-information-content-of-critical-aud
大邓和他的Python
0
我看阿里的年终奖总算发了!
到4月底了,这两天看朋友圈,发现阿里的年终奖终于发了,问了问老同学,也从网上检索了不少信息,基本搞清楚了阿里今年的年终奖情况。近来来阿里一些集团对绩效等级做了较大的调整,以前的旧绩效系统中,绩效分为3.25、3.5、3.75、4和5五个等级,其中4和5是较高绩效等级,较少见。而且之前3.5绩效内部划
公子龙
0
CVPR 2024|大视觉模型的开山之作!无需任何语言数据即可打造大视觉模型
↑ 点击蓝字 关注极市平台作者丨科技猛兽编辑丨极市平台极市导读 本文提出一种序列建模 (sequential modeling) 的方法,不使用任何语言数据,训练大视觉模型。>>加入极市CV技术交流群,走在计算机视觉的最前沿本文目录1 序列建模打造大视觉模型(来自 U
极市平台
1
金融研究(更新) | 使用Python构建关键审计事项的「信息含量」
Tips: 公众号推送后内容只能更改一次,且只能改20字符。如果内容出问题,或者想更新内容, 只能重复推送。为了更好的阅读体验,建议阅读本文博客版, 链接地址https://textdata.cn/blog/2023-01-13-information-content-of-critical-aud
大邓和他的Python
0
字节的跳动职级与薪资(2024年)
上一篇:阿里公布年终奖,P7, 3.5+,22W年终奖,还有35W长期现金激励,真香字节跳动自2012年3月成立以来,已经迅速成长为一个全球性的科技公司。其产品和服务已经遍布全球150多个国家与地区,并且支持超过75种不同的语言。在字节跳动的官方网站上,列出了一系列引人注目的产品和服务,包括但不限于
开发者全社区
0