canal监控mysql变化

2024-06-15 10:12

本文主要是介绍canal监控mysql变化,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

目录

  • 概述
  • 实践
    • 代码
    • 总结
      • 表创建
      • 插入一行数据
      • 更新一行数据
      • 删除一笔数据

概述

   本文测试 canal 监控 mysql 表变化。canal 1.1.7 mysql 8.0.x 版本。

实践

代码

public static void main(String[] args) {// 创建一个 CanalConnector 连接器// username:字符串类型,Canal使用该用户名验证客户端身份// password:字符串类型,Canal使用该密码验证客户端身份CanalConnector canalConnector = CanalConnectors.newSingleConnector(new InetSocketAddress("10.xx.xx.142", 11111), "example", "canal", "canal");try {while (true) {try {// 连接 Canal Server 尝试多次重连canalConnector.connect();break;} catch (Exception e) {System.out.println("重新连接...");Thread.sleep(1000);}}// 订阅数据库表,默认监听所有的数据 库、表、等同于: .*\\..*//canalConnector.subscribe(".*\\..*");// 监听指定的数据库、表canalConnector.subscribe("shop.product");// 回滚到上一次的 batchId,取消已经消费过的日志canalConnector.rollback();// 持续监听 Canal Server 推送的数据,并使用自定义的 CanalEventDownStreamHandler 处理器消费数据while (true) {// 允许指定 batchSize 一次可以获取多条  每次返回的对象为 Message  包含的内容为// batch id 唯一标识// entries 具体的数据对象Message message = canalConnector.getWithoutAck(100);long batchId = message.getId();// 如果没有新数据 则暂停固定时间后  继续获取if (batchId == -1 || message.getEntries().isEmpty()) {Thread.sleep(1000);}else {// 解析 binlog 数据输出详细信息for (CanalEntry.Entry entry : message.getEntries()) {if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {continue;}CanalEntry.RowChange rowChange = null;try {rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());} catch (Exception e) {e.printStackTrace();continue;}String schemaName = entry.getHeader().getSchemaName();String tableName = entry.getHeader().getTableName();CanalEntry.EventType eventType = rowChange.getEventType();System.out.println(String.format("Binlog[%s:%s] schema[%s] table[%s] eventType[%s]",entry.getHeader().getLogfileName(),entry.getHeader().getLogfileOffset(),schemaName,tableName,eventType));for (CanalEntry.RowData rowData : rowChange.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());}}// 确认消费成功canalConnector.ack(batchId);}}}} catch (Exception e) {e.printStackTrace();}finally {canalConnector.disconnect();}}private static void printColumn(List<CanalEntry.Column> columns) {for (CanalEntry.Column column : columns) {System.out.println(column.getName()+" : "+column.getValue() +" update:"+ column.getUpdated());}}

总结

表创建

Binlog[binlog.000002:3153] schema[shop] table[product] eventType[CREATE]

插入一行数据

Binlog[binlog.000002:4484] schema[shop] table[product] eventType[INSERT]
id : 1 update:true
title : 测试 update:true
cover_img : 21 update:true
amout : 11.0 update:true
summary : 11 update:true
detail : 11 update:true
phone : 11 update:true
gmt_create : 2024-06-11 03:11:44 update:true
gmt_modified : 2024-06-11 03:11:44 update:true

更新一行数据

Binlog[binlog.000002:4847] schema[shop] table[product] eventType[UPDATE]
before
id : 1 update:false
title : 测试 update:false
cover_img : 21 update:false
amout : 11.0 update:false
summary : 11 update:false
detail : 11 update:false
phone : 11 update:false
gmt_create : 2024-06-11 03:11:44 update:false
gmt_modified : 2024-06-11 03:11:44 update:false
after
id : 1 update:false
title : 测试99 update:true
cover_img : 21 update:false
amout : 11.0 update:false
summary : 11 update:false
detail : 11 update:false
phone : 11 update:false
gmt_create : 2024-06-11 03:11:44 update:false
gmt_modified : 2024-06-11 03:12:21 update:true

删除一笔数据

Binlog[binlog.000002:5248] schema[shop] table[product] eventType[DELETE]
id : 1 update:false
title : 测试99 update:false
cover_img : 21 update:false
amout : 11.0 update:false
summary : 11 update:false
detail : 11 update:false
phone : 11 update:false
gmt_create : 2024-06-11 03:11:44 update:false
gmt_modified : 2024-06-11 03:12:21 update:false

这篇关于canal监控mysql变化的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1063161

相关文章

MySQL 中的 CAST 函数详解及常见用法

《MySQL中的CAST函数详解及常见用法》CAST函数是MySQL中用于数据类型转换的重要函数,它允许你将一个值从一种数据类型转换为另一种数据类型,本文给大家介绍MySQL中的CAST... 目录mysql 中的 CAST 函数详解一、基本语法二、支持的数据类型三、常见用法示例1. 字符串转数字2. 数字

Mysql实现范围分区表(新增、删除、重组、查看)

《Mysql实现范围分区表(新增、删除、重组、查看)》MySQL分区表的四种类型(范围、哈希、列表、键值),主要介绍了范围分区的创建、查询、添加、删除及重组织操作,具有一定的参考价值,感兴趣的可以了解... 目录一、mysql分区表分类二、范围分区(Range Partitioning1、新建分区表:2、分

MySQL 定时新增分区的实现示例

《MySQL定时新增分区的实现示例》本文主要介绍了通过存储过程和定时任务实现MySQL分区的自动创建,解决大数据量下手动维护的繁琐问题,具有一定的参考价值,感兴趣的可以了解一下... mysql创建好分区之后,有时候会需要自动创建分区。比如,一些表数据量非常大,有些数据是热点数据,按照日期分区MululbU

SQL Server配置管理器无法打开的四种解决方法

《SQLServer配置管理器无法打开的四种解决方法》本文总结了SQLServer配置管理器无法打开的四种解决方法,文中通过图文示例介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的... 目录方法一:桌面图标进入方法二:运行窗口进入检查版本号对照表php方法三:查找文件路径方法四:检查 S

MySQL 删除数据详解(最新整理)

《MySQL删除数据详解(最新整理)》:本文主要介绍MySQL删除数据的相关知识,本文通过实例代码给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考下吧... 目录一、前言二、mysql 中的三种删除方式1.DELETE语句✅ 基本语法: 示例:2.TRUNCATE语句✅ 基本语

MySQL中查找重复值的实现

《MySQL中查找重复值的实现》查找重复值是一项常见需求,比如在数据清理、数据分析、数据质量检查等场景下,我们常常需要找出表中某列或多列的重复值,具有一定的参考价值,感兴趣的可以了解一下... 目录技术背景实现步骤方法一:使用GROUP BY和HAVING子句方法二:仅返回重复值方法三:返回完整记录方法四:

从入门到精通MySQL联合查询

《从入门到精通MySQL联合查询》:本文主要介绍从入门到精通MySQL联合查询,本文通过实例代码给大家介绍的非常详细,需要的朋友可以参考下... 目录摘要1. 多表联合查询时mysql内部原理2. 内连接3. 外连接4. 自连接5. 子查询6. 合并查询7. 插入查询结果摘要前面我们学习了数据库设计时要满

MySQL查询JSON数组字段包含特定字符串的方法

《MySQL查询JSON数组字段包含特定字符串的方法》在MySQL数据库中,当某个字段存储的是JSON数组,需要查询数组中包含特定字符串的记录时传统的LIKE语句无法直接使用,下面小编就为大家介绍两种... 目录问题背景解决方案对比1. 精确匹配方案(推荐)2. 模糊匹配方案参数化查询示例使用场景建议性能优

mysql表操作与查询功能详解

《mysql表操作与查询功能详解》本文系统讲解MySQL表操作与查询,涵盖创建、修改、复制表语法,基本查询结构及WHERE、GROUPBY等子句,本文结合实例代码给大家介绍的非常详细,感兴趣的朋友跟随... 目录01.表的操作1.1表操作概览1.2创建表1.3修改表1.4复制表02.基本查询操作2.1 SE

MySQL中的锁机制详解之全局锁,表级锁,行级锁

《MySQL中的锁机制详解之全局锁,表级锁,行级锁》MySQL锁机制通过全局、表级、行级锁控制并发,保障数据一致性与隔离性,全局锁适用于全库备份,表级锁适合读多写少场景,行级锁(InnoDB)实现高并... 目录一、锁机制基础:从并发问题到锁分类1.1 并发访问的三大问题1.2 锁的核心作用1.3 锁粒度分