如何利用Java实现MySQL的数据变化监听

景梨华

目录

• 1.binlog 简介
• 1.1 什么是 binlog
• 1.2 binlog 的三种格式
• 2. 开启 binlog 并配置 MySQL
• 2.1 检查 binlog 是否开启
• 2.2 修改 MySQL 配置文件（my.cnf 或 my.ini）
• 2.3 验证 binlog 配置
• 3. 使用 Java 监听 binlog
• 3.1 选择工具：Canal
• 3.2 Java 代码监听 binlog
• 4. 代码解析
• 总结

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

1.binlog 简介

1.1 什么是 binlog

binlog（Binary Log） 是 MySQL 记录 DDL（数据定义语言，如

1. CREATE

复制代码

、

1. ALTER

复制代码

）和 DML（数据操作语言，如

1. INSERT

复制代码

、

1. UPDATE

复制代码

、

1. DELETE

复制代码

）的日志文件，它用于：

• 主从复制：MySQL 主库将 binlog 传输到从库，实现数据同步。
• 数据恢复：通过
  1. mysqlbinlog

  复制代码
  工具解析 binlog 恢复数据。
• 数据同步：第三方工具（如 Canal）解析 binlog，进行数据同步。

1.2 binlog 的三种格式

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

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

2. 开启 binlog 并配置 MySQL

2.1 检查 binlog 是否开启

1. SHOW VARIABLES LIKE 'log_bin';

复制代码

如果

1. log_bin

复制代码

值为

1. OFF

复制代码

，说明 binlog 未开启。

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

在

1. [mysqld]

复制代码

部分添加以下内容：

1. server-id=1
2. log-bin=mysql-bin
3. binlog-format=ROW
4. binlog-row-image=FULL
5. expire_logs_days=7

复制代码

重启 MySQL：

1. systemctl restart mysql # Linux
2. net stop mysql && net start mysql # Windows

复制代码

2.3 验证 binlog 配置

执行：

1. SHOW BINARY LOGS;

复制代码

如果有 binlog 文件，如

1. mysql-bin.000001

复制代码

，说明已开启。

3. 使用 Java 监听 binlog

3.1 选择工具：Canal

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

3.2 Java 代码监听 binlog

引入 Maven 依赖

1. <dependencies>
2. <dependency>
3. <groupId>com.alibaba.otter</groupId>
4. <artifactId>canal.client</artifactId>
5. <version>1.1.6</version>
6. </dependency>
7. </dependencies>

复制代码

编写 Java 代码

1. import com.alibaba.otter.canal.client.CanalConnector;
2. import com.alibaba.otter.canal.client.CanalConnectors;
3. import com.alibaba.otter.canal.protocol.CanalEntry;
4. import com.alibaba.otter.canal.protocol.Message;
6. import java.net.InetSocketAddress;
7. import java.util.List;
9. public class BinlogListener {
10. public static void main(String[] args) {
11. // 连接 Canal
12. CanalConnector connector = CanalConnectors.newSingleConnector(
13. new InetSocketAddress("127.0.0.1", 11111),
14. "example", "canal", "canal");
17. try {
18. connector.connect();
19. connector.subscribe(".*\\..*"); // 监听所有库表
20. connector.rollback();
22. while (true) {
23. Message message = connector.getWithoutAck(100); // 获取数据
24. long batchId = message.getId();
25. List<CanalEntry.Entry> entries = message.getEntries();
27. if (batchId != -1 && !entries.isEmpty()) {
28. for (CanalEntry.Entry entry : entries) {
29. if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
30. processEntry(entry);
31. }
32. }
33. }
34. connector.ack(batchId); // 确认消息
35. }
36. } finally {
37. connector.disconnect();
38. }
39. }
41. private static void processEntry(CanalEntry.Entry entry) {
42. try {
43. CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
44. CanalEntry.EventType eventType = rowChange.getEventType();
46. System.out.println("变更表：" + entry.getHeader().getTableName());
47. System.out.println("变更类型：" + eventType);
49. for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
50. if (eventType == CanalEntry.EventType.DELETE) {
51. System.out.println("删除数据：" + rowData.getBeforeColumnsList());
52. } else if (eventType == CanalEntry.EventType.INSERT) {
53. System.out.println("新增数据：" + rowData.getAfterColumnsList());
54. } else {
55. System.out.println("更新前数据：" + rowData.getBeforeColumnsList());
56. System.out.println("更新后数据：" + rowData.getAfterColumnsList());
57. }
58. }
59. } catch (Exception e) {
60. e.printStackTrace();
61. }
62. }
64. }

复制代码

4. 代码解析

1.创建 Canal 连接

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

复制代码

• 1. 127.0.0.1

  复制代码
  ：Canal 服务器地址
• 1. 11111

  复制代码
  ：Canal 端口
• 1. example

  复制代码
  ：Canal 实例
• 1. canal/canal

  复制代码
  ：默认账号密码

2.获取 binlog 变更数据

1. Message message = connector.getWithoutAck(100);

复制代码

1. getWithoutAck(100)

复制代码

：拉取 100 条 binlog 事件。

3.解析 binlog

1. for (CanalEntry.Entry entry : entries) {
2. if (entry.getEntryType() == CanalEntry.EntryType.ROWDATA) {
3. processEntry(entry);
4. }
5. }

复制代码

仅处理

1. ROWDATA

复制代码

类型的变更，忽略事务等其他信息。

4.分类处理 INSERT、UPDATE、DELETE

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

复制代码

总结

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

  复制代码
  、
  1. UPDATE

  复制代码
  、
  1. DELETE

  复制代码
  操作，并解析变更数据。

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

到此这篇关于如何利用Java实现MySQL的数据变化监听的文章就介绍到这了,更多相关Java监听MySQL数据变化内容请搜索晓枫资讯以前的文章或继续浏览下面的相关文章希望大家以后多多支持晓枫资讯！

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！

晴天小盒够

路过，支持一下