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

相关文章

流媒体平台/视频监控/安防视频汇聚EasyCVR播放暂停后视频画面黑屏是什么原因?

视频智能分析/视频监控/安防监控综合管理系统EasyCVR视频汇聚融合平台,是TSINGSEE青犀视频垂直深耕音视频流媒体技术、AI智能技术领域的杰出成果。该平台以其强大的视频处理、汇聚与融合能力,在构建全栈视频监控系统中展现出了独特的优势。视频监控管理系统EasyCVR平台内置了强大的视频解码、转码、压缩等技术,能够处理多种视频流格式,并以多种格式(RTMP、RTSP、HTTP-FLV、WebS

SQL中的外键约束

外键约束用于表示两张表中的指标连接关系。外键约束的作用主要有以下三点: 1.确保子表中的某个字段(外键)只能引用父表中的有效记录2.主表中的列被删除时,子表中的关联列也会被删除3.主表中的列更新时,子表中的关联元素也会被更新 子表中的元素指向主表 以下是一个外键约束的实例展示

基于MySQL Binlog的Elasticsearch数据同步实践

一、为什么要做 随着马蜂窝的逐渐发展,我们的业务数据越来越多,单纯使用 MySQL 已经不能满足我们的数据查询需求,例如对于商品、订单等数据的多维度检索。 使用 Elasticsearch 存储业务数据可以很好的解决我们业务中的搜索需求。而数据进行异构存储后,随之而来的就是数据同步的问题。 二、现有方法及问题 对于数据同步,我们目前的解决方案是建立数据中间表。把需要检索的业务数据,统一放到一张M

如何去写一手好SQL

MySQL性能 最大数据量 抛开数据量和并发数,谈性能都是耍流氓。MySQL没有限制单表最大记录数,它取决于操作系统对文件大小的限制。 《阿里巴巴Java开发手册》提出单表行数超过500万行或者单表容量超过2GB,才推荐分库分表。性能由综合因素决定,抛开业务复杂度,影响程度依次是硬件配置、MySQL配置、数据表设计、索引优化。500万这个值仅供参考,并非铁律。 博主曾经操作过超过4亿行数据

性能分析之MySQL索引实战案例

文章目录 一、前言二、准备三、MySQL索引优化四、MySQL 索引知识回顾五、总结 一、前言 在上一讲性能工具之 JProfiler 简单登录案例分析实战中已经发现SQL没有建立索引问题,本文将一起从代码层去分析为什么没有建立索引? 开源ERP项目地址:https://gitee.com/jishenghua/JSH_ERP 二、准备 打开IDEA找到登录请求资源路径位置

MySQL数据库宕机,启动不起来,教你一招搞定!

作者介绍:老苏,10余年DBA工作运维经验,擅长Oracle、MySQL、PG、Mongodb数据库运维(如安装迁移,性能优化、故障应急处理等)公众号:老苏畅谈运维欢迎关注本人公众号,更多精彩与您分享。 MySQL数据库宕机,数据页损坏问题,启动不起来,该如何排查和解决,本文将为你说明具体的排查过程。 查看MySQL error日志 查看 MySQL error日志,排查哪个表(表空间

综合安防管理平台LntonAIServer视频监控汇聚抖动检测算法优势

LntonAIServer视频质量诊断功能中的抖动检测是一个专门针对视频稳定性进行分析的功能。抖动通常是指视频帧之间的不必要运动,这种运动可能是由于摄像机的移动、传输中的错误或编解码问题导致的。抖动检测对于确保视频内容的平滑性和观看体验至关重要。 优势 1. 提高图像质量 - 清晰度提升:减少抖动,提高图像的清晰度和细节表现力,使得监控画面更加真实可信。 - 细节增强:在低光条件下,抖

MySQL高性能优化规范

前言:      笔者最近上班途中突然想丰富下自己的数据库优化技能。于是在查阅了多篇文章后,总结出了这篇! 数据库命令规范 所有数据库对象名称必须使用小写字母并用下划线分割 所有数据库对象名称禁止使用mysql保留关键字(如果表名中包含关键字查询时,需要将其用单引号括起来) 数据库对象的命名要能做到见名识意,并且最后不要超过32个字符 临时库表必须以tmp_为前缀并以日期为后缀,备份

[MySQL表的增删改查-进阶]

🌈个人主页:努力学编程’ ⛅个人推荐: c语言从初阶到进阶 JavaEE详解 数据结构 ⚡学好数据结构,刷题刻不容缓:点击一起刷题 🌙心灵鸡汤:总有人要赢,为什么不能是我呢 💻💻💻数据库约束 🔭🔭🔭约束类型 not null: 指示某列不能存储 NULL 值unique: 保证某列的每行必须有唯一的值default: 规定没有给列赋值时的默认值.primary key:

SWAP作物生长模型安装教程、数据制备、敏感性分析、气候变化影响、R模型敏感性分析与贝叶斯优化、Fortran源代码分析、气候数据降尺度与变化影响分析

查看原文>>>全流程SWAP农业模型数据制备、敏感性分析及气候变化影响实践技术应用 SWAP模型是由荷兰瓦赫宁根大学开发的先进农作物模型,它综合考虑了土壤-水分-大气以及植被间的相互作用;是一种描述作物生长过程的一种机理性作物生长模型。它不但运用Richard方程,使其能够精确的模拟土壤中水分的运动,而且耦合了WOFOST作物模型使作物的生长描述更为科学。 本文让更多的科研人员和农业工作者