ChunJun实战：如何通过JSON配置实现MySQL Binlog实时同步？

还在为不同数据库之间的数据同步而烦恼，手动编写脚本既耗时又容易出错？今天，我们来聊聊一个高效的解决方案：使用 ChunJun 通过简单的 JSON 配置文件，轻松搞定 MySQL Binlog 的实时捕获与同步。

ChunJun（原 FlinkX）是一款基于 Apache Flink 的分布式数据同步工具，专注于异构数据源之间的高效数据迁移和实时同步。它支持包括 MySQL、Oracle、Hive、Kafka 等在内的多种数据源，提供批流一体的数据处理能力，是构建数据仓库、实现实时数仓同步的理想选择。

下面，我们就通过一个完整的实战示例，带你快速上手。

环境准备：两步即可开局

开始之前，只需确认两件事：

1. 环境就绪：确保 ChunJun 已经正确安装在你的服务器或本地环境。
2. 开启Binlog：这是 MySQL 记录所有数据变更的日志，是实时同步的源头。你需要在 MySQL 的配置文件（通常是 my.cnf 或 my.ini）的 [mysqld] 段落下加入以下配置，并重启 MySQL 服务：

[mysqld]
server-id = 1
log-bin = mysql-bin
binlog-format = ROW

核心：编写任务配置文件

ChunJun 通过一个 JSON 文件来定义同步任务，清晰直观。我们创建一个名为 binlog_stream.json 的文件，其核心结构如下：

{
  "job": {
    "content": [ {
      "reader": {
        "parameter": {
          "schema": "cdc",                    // 要监听的数据库名
          "username": "root",                 // 数据库用户名
          "password": "root",                 // 数据库密码
          "cat": "insert,delete,update",      // 监听的操作类型：增、删、改
          "jdbcUrl": "jdbc:mysql://172.24.0.1:3306/cdc?useSSL=false", // 数据库连接地址
          "host": "172.24.0.1",               // 数据库主机
          "port": 3306,                       // 数据库端口
          "table": [ "chun_test" ],           // 要同步的具体表名
          "splitUpdate": true,                // 将UPDATE操作拆分为变更前和变更后数据
          "pavingData": true                  // 将数据平铺展开，便于后续处理
        },
        "name": "binlogreader"                // 读取器类型：Binlog
      },
      "writer": {
        "parameter": {
          "print": true                       // 为方便演示，将数据打印到控制台
        },
        "name": "streamwriter"                // 写入器类型：标准输出流
      }
    } ],
    "setting": {
      "speed": {
        "bytes": 0,                           // 流量控制，0表示不限速
        "channel": 1                          // 同步通道数，影响并发性能
      }
    }
  }
}

这个配置文件定义了一个最基本的任务：监听 cdc 数据库中 chun_test 表的所有 INSERT, UPDATE, DELETE 操作，并将捕获到的数据变更实时打印出来。

动手实战：验证数据流动

配置文件准备好了，我们搭建一个简单的测试环境，亲眼看看数据是如何流动的。

1. 准备测试数据
首先，在你的 MySQL 数据库中执行以下 SQL，创建测试表并插入初始数据：

-- 1. 创建测试表
CREATE TABLE `chun_test` (
  `id` int(11) NOT NULL,
  `name` varchar(50) DEFAULT NULL,
  `create_time` timestamp NULL DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 2. 插入三条初始数据
INSERT INTO `chun_test` VALUES (1, '张三', '2025-10-01 12:09:33');
INSERT INTO `chun_test` VALUES (2, '李四', '2025-10-02 13:08:22');
INSERT INTO `chun_test` VALUES (3, '王五', '2025-10-03 14:07:11');

2. 启动 ChunJun 同步任务
保存好 binlog_stream.json 文件后，在 ChunJun 的安装目录下执行启动命令：

# 假设配置文件路径为 chunjun-examples/test/binlog_stream.json
bin/chunjun-local.sh -job chunjun-examples/test/binlog_stream.json

当控制台输出连接成功的日志时，说明任务已启动，正在监听 chun_test 表的数据变更。

3. 测试实时同步
保持 ChunJun 任务运行，打开另一个 MySQL 客户端，执行一条新的插入语句：

INSERT INTO `chun_test` VALUES (4, '赵六', '2025-10-04 15:06:00');

此时，你立即可以在 ChunJun 的运行窗口看到新插入的这条记录被捕获并打印出来。这表明基于 Binlog 的实时同步已经成功运行。

进阶技巧：指定位点与断点续传

默认情况下，ChunJun 从当前时刻开始监听。但在实际生产环境中，我们可能需要从某个历史位点开始消费，例如进行数据补全或任务恢复。这在 ChunJun 中很容易实现。

只需在 reader 的 parameter 配置中增加一个 start 节点即可：

"start": {
  "journal-name": "mysql-bin.000001"
}

start 参数为你提供了多种灵活的启动方式：

参数属性	功能
timestamp	从指定的时间戳处开始消费
journal-name	从指定的 Binlog 文件起始处开始消费
position	从指定文件的指定位置（字节偏移量）处开始消费
（默认）	从当前最新的位点开始消费

这个功能赋予了你的数据同步链路“断点续传”的能力，极大地增强了任务的健壮性。

总结

通过上面的实战，我们可以看到，利用 ChunJun 实现 MySQL 数据实时同步的核心优势：

• 配置驱动，简单高效：无需编写复杂代码，通过声明式的 JSON 配置即可定义复杂的同步任务。
• 实时精准，延迟极低：直接解析 MySQL Binlog，能够毫秒级感知数据变更。
• 位点可调，健壮性强：支持从任意历史位点启动，完美支持数据回溯与故障恢复。
• 生态丰富，扩展性强：作为一款优秀的数据同步工具，其能力远不止于 MySQL。它可以轻松对接 Kafka、Hive 等数十种数据源，满足你更复杂的 数据集成 场景。

技术工具的最终目的是提升效率，让开发者更专注于业务逻辑而非底层的数据搬运。希望这篇 ChunJun 的实战指南能帮助你快速上手，解决实际的数据同步难题。如果你在实践过程中遇到问题，或者有更深入的技术思考，欢迎到 云栈社区 与更多开发者交流探讨。