CDC增量数据同步

CDC概述

定义变更数据捕获（Change Data Capture, CDC）是一种实时捕获数据库增量变更（如插入、更新、删除）的技术，并将这些变更事件传递给下游系统进行处理或同步。

作用

实时数据同步：将源数据库的变更实时推送至目标系统（如数据仓库、消息队列、分析平台）。
系统解耦：减少直接依赖，通过异步机制提升系统可靠性。
低延迟处理：避免全量数据扫描，仅处理增量变更，提升效率。
多场景适配：支持数据迁移、容灾备份、业务监控、实时分析等场景。

ESB集成CDC的作用

ESB通过集成CDC技术，为企业提供了一套灵活、高效的数据实时同步与处理方案。无论是跨系统集成、实时分析还是容灾备份，均可通过标准化配置快速实现，同时保障高可靠性与低延迟，助力企业构建敏捷的数据驱动架构。企业服务总线（ESB）通过集成CDC能力，实现以下核心价值：

异构数据库支持： ESB支持多种数据库（DM、HighGo、MySQL、Oracle、OceanBase、OpenGauss、PostgreSQL、SQL Server、Vastbase），统一管理不同数据源的变更捕获。
流程自动化：
- 通过触发器监听CDC事件，自动触发后续业务流程（如数据转换、通知、集成API调用）。
- 支持与自定义脚本、条件分支等组件结合，实现复杂数据处理逻辑。
高可靠性与容错：
- 提供事件处理失败模式（如日志告警后继续执行），保障任务连续性。
- 支持心跳检查、全量快照等机制，确保数据完整性。
资源优化：
- 通过批处理（如批处理大小=4096）、队列控制（如最大队列大小=8192）优化性能。

典型使用场景

实时数据仓库更新：将业务数据库的增量变更实时同步至数据仓库。
跨系统数据同步：在分布式架构中，同步主数据库变更至缓存（如Redis）或微服务数据库。
业务监控与审计：捕获关键表变更，触发告警或记录操作日志。
容灾与备份：实时将生产数据备份至异地容灾中心。

通信链路

CDC ↔ 数据库：
- CDC服务通过数据库协议（如MySQL Connector/J）直连源库，监听日志变更。
CDC ↔ ESB：
- 使用Netty TCP长连接，基于Protobuf序列化传输事件数据。
ESB ↔ 下游系统：
- 支持HTTP/REST、Kafka、JDBC等多种协议输出。

流程（数据同步）配置示例

示例描述：mysql 环境数据库PUBLIC 的表test 同步到testb表中

前提条件

数据库可正常使用。
数据库根据 tongcdc使用说明手册配置。
tongcdc-server 进程启动正常。

运行时配置cdc连接信息

Server安装目录\config\application.yml 配置 cdc 的服务端连接信息。


# 连接cdc的ip
tongcdc.host = ${TONGCDC_IP:127.0.0.1}
# 连接cdc的端口
tongcdc.port = ${TONGCDC_PORT:9090}
# 连接cdc的用户名
tongcdc.username=${TONGCDC_USERNAME:tongcdc}
# 连接cdc的密码
tongcdc.password= ${TONGCDC_PASS:TongCDC@123}
# 连接cdc的工作线程数
tongcdc.nThread=${TONGCDC_THREAD:1}
# 连接cdc的连接超时时间
tongcdc.connectTimeout=${TONGCDC_CONNECT_TIMEOUT:10000}
# 连接cdc的socket超时时间
tongcdc.socketTimeout =${TONGCDC_SOCKET_TIMEOUT:10000}

测试数据库准备

参考tongcdc说明手册，配置支持cdc抽取的环境，然后创建示例表（必须配置主键）

-- PUBLIC.test definition

CREATE TABLE `test` (
  `id` varchar(100) DEFAULT NULL,
  `name` varchar(100) DEFAULT NULL
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;



-- PUBLIC.testb definition

CREATE TABLE `testb` (
  `id2` varchar(100) DEFAULT NULL,
  `name2` varchar(100) DEFAULT NULL,
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;

创建数据库连接

"数据连接" -> 新建-> 数据连接

"选择连接类型" ->"SQL DB" -> “mysql”

填写连接属性配置
连接名称
URL
用户名密码
点击右上角 “测试连接” ，通过后点击“保存按钮”

创建流程

"集成编排" -> 新建 -> "集成流程"

配置入站端点CDC 进行监听

选择创建的 "连接"
填写订阅的schema和表等信息（测试mysql 的schema 为PUBLIC ，表为test）

CDC捕获数据后，触发ESB流程，传递变更事件至后续节点

循环遍历获取变更事件的数据

因为从CDC抽取的数据格式为数组，所以循环处理每一条数据

条件分支

CDC抽取的操作类型分为四种：CREATE、INSERT、UPDATE、DELETE

分别为建表语句，插入操作，更新操作和删除数据操作。

所以使用条件分支判断做不同的数据库操作

INSERT 分支

自定义脚本

通过脚本实现数据格式转换、字段过滤或业务逻辑增强。

对于从cdc接收到的数据（参考：CDC数据抽取事件到ESB的json示例），通过脚本可以解析到最新的表记录数据。

需要获取到 after 部分的变更数据, 通过自定义脚本组装数据库连接步骤的入参

// 脚本解析示例：
// 执行sql: INSERT INTO test (id, name) VALUES('q', 'qweqws');
// cdc 获取到事件，body 内容为 ：5.2 插入数据事件

LOG.info("process {}",  "INSERT");

import com.alibaba.fastjson2.JSONObject;

/*临时存储下一步的入参*/
Map<String,Object> insertBody = new HashMap();

/*获得当次的记录修改内容*/
List after = ${body.after};
for(Object a:after){
    /*name为字段名称，value为插入的值*/
    insertBody.put(a.name,a.value);
}


/*序列化为json字符串向下传递*/
String insertParameters = JSONObject.toJSONString(insertBody);

LOG.info("insert parameters:{}",insertParameters);
// insertParameters -> {"name":"qweqws","id":"q"}
${request.body} = insertParameters;

数据写入出站端点数据库

根据参数进行插入数据库操作。

UPDATE分支

自定义脚本

需要获取到 after 部分的变更数据, 通过自定义脚本组装数据库连接步骤的入参

LOG.info("process {}",  "UPDATE");


import com.alibaba.fastjson2.JSONObject;

/*临时存储下一步的入参*/
Map<String,Object> updateBody = new HashMap();

/*获得当次的记录修改内容*/
List after = ${body.after};
for(Object a:after){
    /*name为字段名称，value为插入的值*/
    updateBody.put(a.name,a.value);
}


/*序列化为json字符串向下传递*/
String updateParameters = JSONObject.toJSONString(updateBody);

LOG.info("update parameters:{}",updateParameters);

${request.body} = updateParameters;

数据更新到出站端点数据库

根据参数进行更新数据库操作。

DELETE分支

自定义脚本

需要获取到 pks 主键信息, 通过主键删除指定的数据记录。

通过自定义脚本组装数据库连接步骤的入参

LOG.info("process {}",  "DELETE");

import com.alibaba.fastjson2.JSONObject;

/*临时存储下一步的入参*/
Map<String,Object> deleteBody = new HashMap();

/*获得主键，表中有多个主键，通过脚本取出删除操作的条件的值*/
List pks = ${body.pks};

for(Object a:pks){
    /*name为字段名称，value为插入的值*/
    deleteBody.put(a.name,a.value);
}

/*序列化为json字符串向下传递*/
String deleteParameters = JSONObject.toJSONString(deleteBody);

LOG.info("delete parameters:{}",deleteParameters);

${request.body} = deleteParameters;

数据删除操作同步到出站端点数据库

根据参数进行删除数据库操作。

CDC配置参数说明

数据源配置

参数名称	类型	必填	说明	默认值	可选值/范围
数据库连接	-	是	选择需要订阅监听的连接配置	-	-
需要订阅的源数据schema	string	是	指定需要订阅的数据库schema	-	-
源数据库字符集编码	string	否	源数据库字符集编码	-	-
密钥	string	否	数据库用户密码加密密钥	-	-

数据订阅配置

参数名称	类型	必填	说明	默认值	可选值/范围
订阅的数据表	string	是	指定需要订阅的数据库表	-	-
订阅的批量大小	int	否	订阅的批量大小	1024	≥1
订阅的批量超时时间	int	否	订阅的批量超时时间	-	≥0
指定源数据库抽取的表schema	string	是	指定源数据库抽取的schema，多个用逗号分隔	-	如: test1,test2
指定源数据库抽取的表	string	是	指定源数据库抽取的表，多个用逗号分隔	-	格式: database:table_01,table_02
数据库 gtid	string	否	数据库gtid设置	-	-

事件处理配置

参数名称	类型	必填	说明	默认值	可选值/范围
事件处理失败模式	string	是	事件处理失败时的处理方式	fail	fail(抛出异常终止), warn(打印警告继续), skip(跳过继续)
是否跳过解析操作类型	string	是	跳过指定的操作类型	none	none(不跳过), i(insert), u(update), d(delete), t(truncate)
是否捕获ddl变更	boolean	是	是否捕获DDL变更	true	true/false

性能配置

参数名称	类型	必填	说明	默认值	可选值/范围
最大队列大小	int	否	阻塞队列可缓存的最大日志事件数量	8192	0-2147483647
批处理大小	int	否	每批次处理事件的最大大小	4096	0-2147483647
读取binlog日志的最大缓存数	int	是	binlog读取器预读缓冲区大小	102400	≥0

高级配置

参数名称	类型	必填	说明	默认值	可选值/范围
是否启用心跳检查	boolean	否	是否启用心跳检查	false	true/false
是否启用全量快照	boolean	否	是否首次全量抽取	false	true/false
客户端的数字ID	string	否	客户端的数字标识	-	-
拉取间隔时间(秒)	int	否	数据拉取间隔时间	10	0-60

CDC数据抽取事件到ESB的json示例：

create建表事件

[
    {
        "eventType": "CREATE",
        "ddlSql": "CREATE TABLE `test` (\n  `id` int NOT NULL,\n  `name` varchar(255) CHARACTER SET utf8mb3 COLLATE utf8mb3_unicode_ci DEFAULT NULL,\n  PRIMARY KEY (`id`)\n) ENGINE=InnoDB DEFAULT CHARSET=utf8mb3 COLLATE=utf8mb3_unicode_ci",
        "opTime": 1719898968277,
        "position": "mysql-bin.000001|18539729",
        "txId": null,
        "txSeq": 1,
        "pks": null,
        "oldPks": null,
        "before": null,
        "after": null
    }
]

插入数据事件

[
    {
        "eventType": "INSERT",
        "ddlSql": null,
        "opTime": 1719899155000,
        "position": "mysql-bin.000001|18541676",
        "txId": "file=mysql-bin.000001,pos=18541543",
        "txSeq": 1,
        "pks": [
            {
                "name": "id",
                "jdbcType": 4,
                "index": 1,
                "size": 10,
                "digit": 0,
                "value": 1,
                "null": false,
                "update": true,
                "pk": true
            }
        ],
        "oldPks": null,
        "before": null,
        "after": [
            {
                "name": "id",
                "jdbcType": 4,
                "index": 1,
                "size": 10,
                "digit": 0,
                "value": 1,
                "null": false,
                "update": true,
                "pk": true
            },
            {
                "name": "name",
                "jdbcType": 12,
                "index": 2,
                "size": 255,
                "digit": 0,
                "value": "1name",
                "null": false,
                "update": true,
                "pk": false
            }
        ]
    }
]

更新数据事件

[
    {
        "eventType": "UPDATE",
        "ddlSql": null,
        "opTime": 1719899300000,
        "position": "mysql-bin.000001|18543862",
        "txId": "file=mysql-bin.000001,pos=18543720",
        "txSeq": 1,
        "pks": [
            {
                "name": "id",
                "jdbcType": 4,
                "index": 1,
                "size": 10,
                "digit": 0,
                "value": 1,
                "null": false,
                "update": false,
                "pk": true
            }
        ],
        "oldPks": null,
        "before": [
            {
                "name": "id",
                "jdbcType": 4,
                "index": 1,
                "size": 10,
                "digit": 0,
                "value": 1,
                "null": false,
                "update": false,
                "pk": true
            },
            {
                "name": "name",
                "jdbcType": 12,
                "index": 2,
                "size": 255,
                "digit": 0,
                "value": "1name",
                "null": false,
                "update": false,
                "pk": false
            }
        ],
        "after": [
            {
                "name": "id",
                "jdbcType": 4,
                "index": 1,
                "size": 10,
                "digit": 0,
                "value": 1,
                "null": false,
                "update": false,
                "pk": true
            },
            {
                "name": "name",
                "jdbcType": 12,
                "index": 2,
                "size": 255,
                "digit": 0,
                "value": "1name-up",
                "null": false,
                "update": true,
                "pk": false
            }
        ]
    }
]

删除数据事件

[
    {
        "eventType": "DELETE",
        "ddlSql": null,
        "opTime": 1719899346000,
        "position": "mysql-bin.000001|18544797",
        "txId": "file=mysql-bin.000001,pos=18544664",
        "txSeq": 1,
        "pks": [
            {
                "name": "id",
                "jdbcType": 4,
                "index": 1,
                "size": 10,
                "digit": 0,
                "value": 1,
                "null": false,
                "update": false,
                "pk": true
            }
        ],
        "oldPks": null,
        "before": [
            {
                "name": "id",
                "jdbcType": 4,
                "index": 1,
                "size": 10,
                "digit": 0,
                "value": 1,
                "null": false,
                "update": false,
                "pk": true
            },
            {
                "name": "name",
                "jdbcType": 12,
                "index": 2,
                "size": 255,
                "digit": 0,
                "value": "1name-up",
                "null": false,
                "update": false,
                "pk": false
            }
        ],
        "after": null
    }
]

更新主键事件

先DELETE再INSERT

[
    {
        "eventType": "DELETE",
        "ddlSql": null,
        "opTime": 1719901122000,
        "position": "mysql-bin.000001|20925843",
        "txId": "file=mysql-bin.000001,pos=20925701",
        "txSeq": 1,
        "pks": [
            {
                "name": "id",
                "jdbcType": 4,
                "index": 1,
                "size": 10,
                "digit": 0,
                "value": 3,
                "null": false,
                "update": false,
                "pk": true
            }
        ],
        "oldPks": null,
        "before": [
            {
                "name": "id",
                "jdbcType": 4,
                "index": 1,
                "size": 10,
                "digit": 0,
                "value": 3,
                "null": false,
                "update": false,
                "pk": true
            },
            {
                "name": "name",
                "jdbcType": 12,
                "index": 2,
                "size": 255,
                "digit": 0,
                "value": "v",
                "null": false,
                "update": false,
                "pk": false
            }
        ],
        "after": null
    },
    {
        "eventType": "INSERT",
        "ddlSql": null,
        "opTime": 1719901122000,
        "position": "mysql-bin.000001|20925843",
        "txId": "file=mysql-bin.000001,pos=20925701",
        "txSeq": 2,
        "pks": [
            {
                "name": "id",
                "jdbcType": 4,
                "index": 1,
                "size": 10,
                "digit": 0,
                "value": 4,
                "null": false,
                "update": true,
                "pk": true
            }
        ],
        "oldPks": null,
        "before": null,
        "after": [
            {
                "name": "id",
                "jdbcType": 4,
                "index": 1,
                "size": 10,
                "digit": 0,
                "value": 4,
                "null": false,
                "update": true,
                "pk": true
            },
            {
                "name": "name",
                "jdbcType": 12,
                "index": 2,
                "size": 255,
                "digit": 0,
                "value": "va",
                "null": false,
                "update": true,
                "pk": false
            }
        ]
    }
]

CDC概述​

ESB集成CDC的作用​

典型使用场景​

通信链路​

流程（数据同步）配置示例​

前提条件​

运行时配置cdc连接信息​

测试数据库准备​

创建数据库连接​

创建流程​

配置入站端点CDC 进行监听​

循环遍历获取变更事件的数据​

条件分支​

INSERT 分支​

自定义脚本​

数据写入出站端点数据库​

UPDATE分支​

自定义脚本​

数据更新到出站端点数据库​

DELETE分支​

自定义脚本​

数据删除操作同步到出站端点数据库​

CDC配置参数说明​

数据源配置​

数据订阅配置​

事件处理配置​

性能配置​

高级配置​

CDC数据抽取事件到ESB的json示例：​

create建表事件​

插入数据事件​

更新数据事件​

删除数据事件​

更新主键事件​

CDC概述

ESB集成CDC的作用

典型使用场景

通信链路

流程（数据同步）配置示例

前提条件

运行时配置cdc连接信息

测试数据库准备

创建数据库连接

创建流程

配置入站端点CDC 进行监听

循环遍历获取变更事件的数据

条件分支

INSERT 分支

自定义脚本

数据写入出站端点数据库

UPDATE分支

自定义脚本

数据更新到出站端点数据库

DELETE分支

自定义脚本

数据删除操作同步到出站端点数据库

CDC配置参数说明

数据源配置

数据订阅配置

事件处理配置

性能配置

高级配置

CDC数据抽取事件到ESB的json示例：

create建表事件

插入数据事件

更新数据事件

删除数据事件

更新主键事件