📚 课程目录第4章数据源连接器之PostgreSQL CDC

4、数据源连接器之PostgreSQL CDC：PostgreSQL逻辑复制原理、Flink PostgreSQL CDC Connector实战

这一章我们聊聊PostgreSQL CDC。说实话，CDC（Change Data Capture）在实时数据同步里是个绕不开的话题。而PostgreSQL的CDC方案，我个人认为是最优雅的之一——它不需要像某些数据库那样装额外的Agent，也不需要频繁地轮询表。它的核心就是逻辑复制。

你想想看，一个数据库每天产生几百万条变更，如果全靠轮询，不仅慢，还会给数据库带来不小的压力。逻辑复制就不一样了，它让数据库主动把变更“推”出来。嗯，这就是我们这节课要讲的核心。

核心要点：PostgreSQL CDC 的本质是让数据库成为变更事件的“生产者”，Flink 作为“消费者”实时消费这些事件。整个过程对业务库的侵入性极低。

4.1 PostgreSQL逻辑复制原理

逻辑复制，说白了就是PostgreSQL把每个数据变更（INSERT、UPDATE、DELETE）都记录成一种特殊的日志——WAL（Write-Ahead Log）。然后，一个叫“输出插件”的东西把这些日志解析成我们能理解的格式，再通过网络发送给订阅者。

我在项目中遇到过不少新手，一上来就问：“是不是要改PostgreSQL的配置？”其实只需要改几个参数，而且都是在线修改，不需要重启。下面这张图能帮你快速理解整个流程：

整个链条其实就三步：

第一步：业务表发生数据变更，PostgreSQL 把变更写入 WAL 日志。
第二步：输出插件（pgoutput）把 WAL 中的二进制数据解析成逻辑变更事件。
第三步：通过 Publication/Subscription 机制，把事件推送给订阅者（也就是 Flink CDC Connector）。

这里有个关键概念——逻辑复制槽。它就像个“书签”，记录了消费者已经消费到哪个位置了。如果消费者挂了，复制槽会保留未消费的 WAL 日志。我曾经遇到过因为消费者长时间宕机，导致复制槽积压了大量 WAL，差点把磁盘撑爆。嗯，这个坑后面会细说。

4.2 PostgreSQL CDC 配置准备

在写代码之前，得先把 PostgreSQL 的“开关”打开。我个人习惯把这些配置写在 postgresql.conf 里：

# 开启 WAL 逻辑复制（必须）
wal_level = logical

# 复制槽相关（建议）
max_replication_slots = 10
max_wal_senders = 10

# 其他推荐配置
track_commit_timestamp = on

改完之后，执行 SELECT pg_reload_conf(); 就能生效，不需要重启。然后创建发布（Publication）：

-- 创建发布，指定要同步的表
CREATE PUBLICATION flink_pub FOR TABLE t_user, t_order;

-- 或者同步所有表（不推荐生产环境用）
-- CREATE PUBLICATION flink_pub FOR ALL TABLES;

小提示：创建发布时，建议明确指定表名。用 FOR ALL TABLES 虽然省事，但后续新增的表会自动加入发布，可能会同步到不该同步的数据。我在生产环境吃过这个亏。

4.3 Flink PostgreSQL CDC Connector 实战

好了，配置搞定了，接下来就是写代码。Flink CDC Connector 用起来其实很简单，核心就是构建一个 SourceFunction。我直接上代码：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import com.ververica.cdc.connectors.postgres.PostgreSQLSource;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;

public class PostgresCdcExample {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 构建 PostgreSQL CDC Source
        PostgreSQLSource<String> source = PostgreSQLSource.<String>builder()
            .hostname("192.168.1.100")
            .port(5432)
            .database("mydb")           // 数据库名
            .schemaList("public")        // schema 列表
            .tableList("public.t_user")  // 要监听的表
            .username("cdc_user")
            .password("cdc_pass")
            .deserializer(new JsonDebeziumDeserializationSchema()) // 输出 JSON
            .slotName("flink_slot")      // 逻辑复制槽名称
            .decodingPluginName("pgoutput") // 输出插件
            .build();

        env.addSource(source)
            .print();

        env.execute("PostgreSQL CDC Sync");
    }
}

这段代码跑起来之后，只要 t_user 表有 INSERT、UPDATE、DELETE，控制台就会打印出对应的 JSON 事件。比如一条 INSERT 会输出类似这样的数据：

{
  "before": null,
  "after": {
    "id": 1001,
    "name": "张三",
    "age": 28
  },
  "source": {
    "db": "mydb",
    "table": "t_user",
    "lsn": 123456789
  },
  "op": "c",  // c=create, u=update, d=delete
  "ts_ms": 1700000000000
}

这里要注意几个参数：

slotName：每个消费者要有唯一的复制槽名称。如果多个作业用同一个槽名，会冲突。
decodingPluginName：PostgreSQL 14 及以上推荐用 pgoutput，兼容性最好。
tableList：格式是 schema.table，支持通配符，比如 public.*。

4.4 避坑指南与最佳实践

这部分是我最想分享的。CDC 用起来简单，但坑也不少。我一个个说：

⚠️ 复制槽积压问题：我曾经有一个 Flink 作业因为网络抖动挂了 2 小时，结果复制槽积压了 5GB 的 WAL 日志。PostgreSQL 不会自动清理未消费的 WAL，磁盘差点爆了。解决方案：监控复制槽的延迟，设置告警；同时给复制槽设置一个最大保留大小。

其他常见问题：

问题	原因	解决方案
Schema 变更导致作业失败	表结构改了，但 CDC 事件格式不匹配	使用 `includeSchemaChanges=true` 捕获 DDL 事件
大事务导致 OOM	一个事务更新了 100 万行，Flink 缓存了所有事件	设置 `debezium.max.queue.size` 限制队列大小
时区问题	PostgreSQL 的 TIMESTAMP 不带时区	统一使用 UTC，或者在反序列化时指定时区
主键变更	UPDATE 事件依赖主键定位 before 数据	确保表有主键；无主键表用 REPLICA IDENTITY FULL

关于性能调优，我建议关注这几个参数：

debezium.snapshot.fetch.size：全量快照阶段每次拉取的行数，默认 1024。表数据量大时可以调到 5000。
debezium.max.batch.size：每次从复制槽拉取的最大事件数，默认 2048。根据你的处理能力调整。
heartbeat.interval.ms：心跳间隔，默认 0（不发送）。建议设为 5000，避免长时间无变更时连接超时。

个人经验：如果你同步的表数据量很大（比如上亿行），第一次启动 CDC 时会先做全量快照。这个阶段可能会很慢。我建议先手动做一次全量同步，然后再启动 CDC 只同步增量。Flink CDC 2.0 之后支持动态加表，也能缓解这个问题。

4.5 总结一下

PostgreSQL CDC 的核心就是逻辑复制。它让数据库主动推送变更，而不是 Flink 去轮询。配置上只需要改几个参数、创建一个发布，剩下的交给 Flink CDC Connector 就行。

嗯，这一章的内容就到这里。代码和配置都很直接，但生产环境里的坑不少。复制槽积压、大事务 OOM、Schema 变更——这些我都踩过。希望你能少走弯路。