Spring Boot 集成 Debezium 实现 PostgreSQL 增量同步

检查 relreplident 字段，若为 f 则说明配置成功。

应用开发

在 Spring Boot 项目中引入 Debezium 相关依赖。pom.xml 中主要包含 API、嵌入式引擎及 PostgreSQL 连接器：

<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-api</artifactId>
    <version>${debezium.version}</version>
</dependency>
<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-embedded</artifactId>
    <version>${debezium.version}</version>
</dependency>
<dependency>
    <groupId>io.debezium</groupId>
    <artifactId>debezium-connector-postgres</artifactId>
    <version>${debezium.version}</version>
</dependency>

配置类负责定义连接器参数，包括数据库地址、用户名、服务器 ID 及监听表列表。注意 offset.storage 设置为临时文件路径，生产环境建议替换为 Kafka Connect 存储：

@Bean
public io.debezium.config.Configuration customerConnector(Environment env) throws IOException {
    File offsetStorageTempFile = File.createTempFile("offsets_", ".dat");
    return io.debezium.config.Configuration.create()
        .with("name", "customer_postgres_connector")
        .with("connector.class", "io.debezium.connector.postgresql.PostgresConnector")
        .with("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore")
        .with("offset.storage.file.filename", offsetStorageTempFile.getAbsolutePath())
        .with("database.hostname", env.getProperty("customer.datasource.host"))
        .with("database.port", env.getProperty("customer.datasource.port"))
        .with("database.user", env.getProperty("customer.datasource.username"))
        .with("database.password", env.getProperty("customer.datasource.password"))
        .with("database.dbname", env.getProperty("customer.datasource.database"))
        .with("database.server.id", "10181")
        .with("database.server.name", "customer-postgres-db-server")
        .with("table.include.list", "public.t_user")
        .with("publication.autocreate.mode", "filtered")
        .with("plugin.name", "pgoutput")
        .with("slot.name", "dbz_customerdb_listener")
        .build();
}

监听器组件利用 DebeziumEngine 异步处理变更事件。当数据库发生写入时，引擎会回调 handleChangeEvent 方法。这里需要注意，READ 操作通常在初始化时触发，实际业务逻辑应过滤掉该状态，仅处理 CREATE、UPDATE 或 DELETE：

private void handleChangeEvent(RecordChangeEvent<SourceRecord> sourceRecordRecordChangeEvent) {
    SourceRecord sourceRecord = sourceRecordRecordChangeEvent.record();
    // 解析 Key 和 Value，提取变更数据
    Object sourceRecordChangeValue = (Struct) sourceRecord.value();
    // 此处可接入业务逻辑处理 payload
}

启动应用前，在 application.properties 中填入数据库连接信息：

customer.datasource.host=localhost
customer.datasource.port=30028
customer.datasource.database=test_db
customer.datasource.username=test1
customer.datasource.password=test123

验证与调试

启动 Spring Boot 应用后，向 t_user 表插入一条测试数据：

INSERT INTO public.t_user(id, "name", age) VALUES(1, 'harries', 18);

观察控制台日志，可以看到类似以下的输出，其中 op=c 代表创建操作，after 字段包含了新插入的数据快照：

2024-04-07 14:22:01.621 INFO ... i.d.connector.common.BaseSourceTask : 1 records sent during previous 00:00:42.015
2024-04-07 14:22:01.622 INFO ... c.et.postgres.listener.DebeziumListener : Key = Struct{id=1}, Value = Struct{after=Struct{name=harries,age=18}...}

这表明增量同步链路已打通，后续可根据实际需求对接消息队列或下游系统。