MySQL数据变化监听的实现方案

更新时间：2025年02月10日 10:33:31 作者：Huooya

在高并发和大数据环境下,实时获取MySQL数据库的增量变化对数据同步、数据分析、缓存更新等场景至关重要,MySQL的binlog（Binary Log）记录了数据库的所有变更,可以用来实现增量数据监听,本文将介绍如何利用binlog监听MySQL数据增量,并提供基 Java的Canal实现示例

1. binlog 简介

1.1 什么是 binlog？

binlog（Binary Log） 是 MySQL 记录 DDL（数据定义语言，如 CREATE、ALTER）和 DML（数据操作语言，如 INSERT、UPDATE、DELETE）的日志文件，它用于：

主从复制：MySQL 主库将 binlog 传输到从库，实现数据同步。
数据恢复：通过 mysqlbinlog 工具解析 binlog 恢复数据。
数据同步：第三方工具（如 Canal）解析 binlog，进行数据同步。

1.2 binlog 的三种格式

binlog 格式	说明
STATEMENT	记录 SQL 语句本身
ROW	记录行数据变更（推荐）
MIXED	结合前两者，MySQL 自动判断

由于 ROW 格式能提供精确的行级别变更信息，因此推荐使用它。

2. 开启 binlog 并配置 MySQL

2.1 检查 binlog 是否开启

SHOW VARIABLES LIKE 'log_bin';

如果 log_bin 值为 OFF，说明 binlog 未开启。

2.2 修改 MySQL 配置文件（my.cnf 或 my.ini）

在 [mysqld] 部分添加以下内容：

server-id=1
log-bin=mysql-bin
binlog-format=ROW
binlog-row-image=FULL
expire_logs_days=7

重启 MySQL：

systemctl restart mysql  # Linux
net stop mysql && net start mysql  # Windows

2.3 验证 binlog 配置

执行：

SHOW BINARY LOGS;

如果有 binlog 文件，如 mysql-bin.000001，说明已开启。

3. 使用 Java 监听 binlog

3.1 选择工具：Canal

阿里巴巴开源的 Canal 可以模拟 MySQL 从库协议，解析 binlog 并实时推送增量数据。

3.2 Java 代码监听 binlog

引入 Maven 依赖

<dependencies>
    <dependency>
        <groupId>com.alibaba.otter</groupId>
        <artifactId>canal.client</artifactId>
        <version>1.1.6</version>
    </dependency>
</dependencies>

编写 Java 代码

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;

import java.net.InetSocketAddress;
import java.util.List;

public class BinlogListener {
    public static void main(String[] args) {
        // 连接 Canal
        CanalConnector connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress("127.0.0.1", 11111), 
                "example", "canal", "canal");
        

        try {
            connector.connect();
            connector.subscribe(".*\\..*"); // 监听所有库表
            connector.rollback();
    
            while (true) {
                Message message = connector.getWithoutAck(100); // 获取数据
                long batchId = message.getId();
                List<CanalEntry.Entry> entries = message.getEntries();
    
                if (batchId != -1 && !entries.isEmpty()) {
                    for (CanalEntry.Entry entry : entries) {
                        if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
                            processEntry(entry);
                        }
                    }
                }
                connector.ack(batchId); // 确认消息
            }
        } finally {
            connector.disconnect();
        }
    }
    
    private static void processEntry(CanalEntry.Entry entry) {
        try {
            CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            CanalEntry.EventType eventType = rowChange.getEventType();
    
            System.out.println("变更表：" + entry.getHeader().getTableName());
            System.out.println("变更类型：" + eventType);
    
            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                if (eventType == CanalEntry.EventType.DELETE) {
                    System.out.println("删除数据：" + rowData.getBeforeColumnsList());
                } else if (eventType == CanalEntry.EventType.INSERT) {
                    System.out.println("新增数据：" + rowData.getAfterColumnsList());
                } else {
                    System.out.println("更新前数据：" + rowData.getBeforeColumnsList());
                    System.out.println("更新后数据：" + rowData.getAfterColumnsList());
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

}

4. 代码解析

创建 Canal 连接

CanalConnector connector = CanalConnectors.newSingleConnector(
    new InetSocketAddress("127.0.0.1", 11111), 
    "example", "canal", "canal");

- 127.0.0.1：Canal 服务器地址
- 11111：Canal 端口
- example：Canal 实例
- canal/canal：默认账号密码
获取 binlog 变更数据

Message message = connector.getWithoutAck(100);

- getWithoutAck(100)：拉取 100 条 binlog 事件。
解析 binlog

for (CanalEntry.Entry entry : entries) {
    if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
        processEntry(entry);
    }
}

仅处理 ROWDATA 类型的变更，忽略事务等其他信息。
分类处理 INSERT、UPDATE、DELETE

if (eventType == CanalEntry.EventType.DELETE) {
    System.out.println("删除数据：" + rowData.getBeforeColumnsList());
} else if (eventType == CanalEntry.EventType.INSERT) {
    System.out.println("新增数据：" + rowData.getAfterColumnsList());
} else {
    System.out.println("更新前数据：" + rowData.getBeforeColumnsList());
    System.out.println("更新后数据：" + rowData.getAfterColumnsList());
}

总结

MySQL binlog 记录数据库变更，可用于监听增量数据。
Canal 作为 MySQL 从库解析 binlog，实现数据同步。
Java 代码示例 展示如何用 Canal 监听 INSERT、UPDATE、DELETE 操作，并解析变更数据。

这种方案适用于 分布式数据同步、缓存一致性 和 数据变更通知，是实时数据处理的重要手段。

以上就是MySQL数据变化监听的实现方案的详细内容，更多关于MySQL数据变化监听的资料请关注脚本之家其它相关文章！

您可能感兴趣的文章:

MySQL中int(10)和int(11)的区别详解
本文主要介绍了MySQL中int(10)和int(11)的区别详解，文中通过示例代码介绍的非常详细，对大家的学习或者工作具有一定的参考学习价值，需要的朋友们下面随着小编来一起学习学习吧
2023-03-03
MySQL中UPDATE与DELETE语句的使用教程
这篇文章主要介绍了MySQL中UPDATE与DELETE语句的使用教程,是MySQL入门学习中的基础知识,需要的朋友可以参考下
2015-12-12
提升MYSQL查询效率的10个SQL语句优化技巧
MySQL数据库执行效率对程序的执行速度有很大的影响，有效的处理优化数据库是非常有用的。尤其是大量数据需要处理的时候
2018-03-03
Linux下rpm方式安装mysql教程
这篇文章主要为大家详细介绍了Linux下rpm方式安装mysql教程,具有一定的参考价值，感兴趣的小伙伴们可以参考一下
2016-10-10
MySQL表的重命名字段添加及字段属性修改操作语法
这篇文章主要为大家介绍了MySQL表的重命名字段添加及字段属性修改语法，有需要的朋友可以借鉴参考下，希望能够有所帮助，祝大家多多进步，早日升职加薪
2023-05-05
浅谈MySQL的性能优化
这篇文章主要介绍了浅谈MySQL的性能优化,MySQL性能优化是通过对数据库的配置、查询优化以及索引优化等手段提高数据库的响应速度和处理能力,本文从多个层面对mysql性能优化进行了小结,需要的朋友可以参考下
2023-08-08
MySQL 使用索引扫描进行排序
mysql可以使用同一个索引既满足排序，又用于查找行，因此，如果可能，设计索引时应该尽可能地同时满足这两种任务，这样是最好的。本文将介绍如何利用索引来进行排序
2021-06-06
MySQL下200GB大表备份的操作(利用传输表空间解决停服发版表备份问题)
这篇文章主要介绍了MySQL下200GB大表备份的操作(利用传输表空间解决停服发版表备份问题),本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下
2025-04-04
MySQL Order by 语句用法与优化详解
Order by语句是用来排序的，经常我们会使用到Order by来进行排序，下面我给大家来讲讲Order by用法与优化排序，有需要的同学可参考
2013-06-06
MySQL Json类型字段IN查询分组优化
这篇文章主要为大家介绍了MySQL Json类型字段IN查询分组优化,有需要的朋友可以借鉴参考下,希望能够有所帮助,祝大家多多进步,早日升职加薪
2023-08-08