如何用Debezium与Kafka实现PostgreSQL CDC实时数据同步

PostgreSQL 从逻辑解码能力开放之后，把数据库变更精确地推向消息队列成为一条非常实用的路径。Debezium 作为 Kafka Connect 的 CDC 连接器，负责从 PostgreSQL 的复制槽读取变更并标准化成 Kafka 消息，这样下游不管是数仓、缓存还是微服务，都可以低延迟地感知业务表的变化。整个链路并不要求改写业务代码，也不依赖触发器，核心是把数据库本身已经产生的 WAL 记录转换成可消费的事件流。

不过实际落地时，要在配置、监控和消费模型上做好设计，否则容易出现 WAL 积压、连接器重启后位点丢失、Topic 数量失控等问题。先理解 PostgreSQL 的逻辑解码机制，再进入连接器配置和消费端实践，可以少走很多弯路。

一、PostgreSQL 逻辑解码：CDC 的能力底座

PostgreSQL 原生的 CDC 不是靠扫描表实现，而是由 WAL 驱动的。WAL 原本用于崩溃恢复，所有已提交事务都会先写入 WAL，在高可用环境中再同步到备库。开启 logical 级别后，PostgreSQL 可以通过 pgoutput 插件把 WAL 中的物理记录解码为逻辑行变更，内容包括事务 ID、LSN、表名以及每行数据的前后镜像。Debezium 正是借助这个输出插件读取变更流，从而避免自己解析二进制 WAL。

要让数据库具备逻辑解码条件，首先需要把 wal_level 设置为 logical，同时为复制连接预留足够的 max_wal_senders 和 max_replication_slots。复制槽是 CDC 的核心对象，它保存了消费者已经读取到的 LSN 位置，即使连接器暂时离线，数据库也不会提前清理对应 WAL 段。下面的配置和发布创建语句可以作为初始环境准备：

-- postgresql.conf 或 ALTER SYSTEM
ALTER SYSTEM SET wal_level = logical;
ALTER SYSTEM SET max_replication_slots = 8;
ALTER SYSTEM SET max_wal_senders = 8;

-- 修改 wal_level 后需要重启数据库
-- 重启完成后执行
CREATE PUBLICATION debezium_publication FOR TABLE public.orders, public.users;

需要注意，复制槽带来安全水位的同时也带来风险：如果消费者长期不消费，WAL 会持续膨胀，甚至占满磁盘。因此生产环境必须监控 pg_replication_slots 中的 active、restart_lsn 以及 confirmed_flush_lsn 与当前 WAL 位置的差距。发布端 publication 可以指定需要订阅的表，也可以后续通过 ALTER PUBLICATION 动态增减对象，但增减对象时需要同步评估下游消费逻辑。

二、Kafka Connect 上注册 Debezium PostgreSQL 连接器

Debezium 运行在 Kafka Connect 框架中，本质是一个 Connector。部署时建议把 debezium-connector-postgres 的 jar 和插件依赖放到 Kafka Connect 插件目录，然后通过 plugin.path 指定。Connect 通常以 distributed 模式运行，因为它支持多节点负载均衡、任务重分配和 REST API 注册连接器。启动后，连接器实例会建立与 PostgreSQL 的复制连接，并开始从复制槽读取变更。

注册连接器一般通过 Kafka Connect REST 接口提交 JSON 配置。下面是一份可直接参考的 PostgreSQL CDC 配置，包含数据库连接、发布端、复制槽以及消息转换器设置：

{
  "name": "postgres-orders-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "127.0.0.1",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "debezium123",
    "database.dbname": "orders_db",
    "database.server.name": "pg_db",
    "table.include.list": "public.orders,public.users",
    "plugin.name": "pgoutput",
    "publication.name": "debezium_publication",
    "slot.name": "debezium_slot",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter.schemas.enable": "false"
  }
}

这份配置中的 plugin.name 指定使用 PostgreSQL 原生 pgoutput，它比旧版 decoderbufs 部署更简单，不需要额外编译。key.converter.schemas.enable 和 value.converter.schemas.enable 设为 false 后，Kafka 消息体只保留 JSON 数据本身，去掉 Schema 字段，能减少消息体积。table.include.list 限制了只订阅部分表，可以避免生成过多 Topic。

三、Kafka 中的变更消息长什么样

连接器运行后，每个已订阅表会对应一个 Kafka Topic。默认 Topic 名称遵循数据库服务器名.schema.表名 的规则，例如 pg_db.public.orders。消息的 key 在默认配置下是主键值，value 是完整变更结构。使用主键作为 key 有一个重要好处：同一主键的变更会被有序地写入同一分区，对下游实现幂等和顺序处理非常关键。

{
  "schema": { },
  "payload": {
    "before": null,
    "after": {
      "id": 1001,
      "customer": "小张",
      "amount": 299.50,
      "status": "PAID"
    },
    "source": {
      "version": "2.x",
      "connector": "postgresql",
      "name": "pg_db",
      "ts_ms": 1718000000000,
      "snapshot": "false",
      "db": "orders_db",
      "schema": "public",
      "table": "orders",
      "txId": 551,
      "lsn": 289123456789
    },
    "op": "c",
    "ts_ms": 1718000000123,
    "transaction": null
  }
}

上面是一条插入操作的典型消息。op 字段取值 c 表示 create，u 表示 update，d 表示 delete，r 表示快照读取。插入时 before 为 null，after 携带完整新行；更新时 before 和 after 都可能出现；删除时 after 为 null。source 块中的 lsn 和 txId 是 PostgreSQL 侧的事务元数据，能帮助追踪位点和恢复进度。快照模式下产生的 op 为 r，下游需要与增量数据统一处理。

四、消费端如何处理变更事件

下游消费 CDC 消息最直接的方式是用 Kafka 客户端订阅对应 Topic。无论是 Java、Go 还是 Python 客户端，核心逻辑都是监听多个分区并处理 op 类型。下面先用命令行消费者快速验证消息是否正常到达：

kafka-console-consumer.sh --bootstrap-server localhost:9092 \
  --topic pg_db.public.orders \
  --from-beginning

生产场景更常见的是使用 Kafka Streams、Flink 或自定义消费者接入。需要注意 Kafka 只能保证分区内有序，跨分区没有全局顺序。如果业务要求全表顺序，要么让 Topic 只有一个分区，要么在消费端根据事务标识和 LSN 做排序。通常事务内多表变更需要利用 transaction.id 或自定义缓冲区完成聚合。

消费位移管理是另一个重要问题。如果关闭自动提交，消费者必须在自己的事务提交后同步提交 Kafka offset，否则重复消费时会重复执行相同的数据库写入，造成数据不一致。推荐把目标写入和 Offset 提交放在同一个本地事务中，或者让下游具备天然幂等性，例如使用唯一键去重或基于事件版本号判断覆盖。

五、关键调优与故障排查

CDC 管道运行一段时间后，排查问题大多集中在延迟变大、连接器频繁重启、WAL 积压这几类。Kafka Connect 的 heartbeat.interval.ms 和 poll.interval.ms 值得重点调整，前者维持复制连接心跳，后者控制拉取频率。特别是网络有 NAT 或防火墙时，心跳过大会导致连接被中间设备静默断开。

{
  "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
  "database.server.name": "pg_db",
  "heartbeat.interval.ms": "5000",
  "poll.interval.ms": "1000",
  "max.batch.size": "2048",
  "max.queue.size": "8192",
  "snapshot.mode": "initial",
  "replica.identity": "full",
  "table.include.list": "public.orders",
  "slot.name": "debezium_slot",
  "publication.name": "debezium_publication"
}

另外，replica.identity 默认只记录主键和变更列，如果下游需要完整的旧值，应将其设置为 full。这会增加 WAL 体积，但在审计、对账等场景必须如此。大事务场景可以适当提升 max.batch.size 和 max.queue.size，但不要盲目调大，它们会直接影响内存占用和 Kafka 消息延迟。

最后一个容易被忽略的点是 PostgreSQL 用户权限。Debezium 使用的数据库账号必须具备 REPLICATION 权限和订阅表的 SELECT 权限。初次启动快照时如果权限不足，连接器会陷入失败重试。通过查询 pg_stat_replication 可以看到连接器连接状态，通过 pg_replication_slots 可以看到复制槽的滞后字节数。把这两类指标接入监控，就能在 WAL 膨胀影响数据库可用性之前触发告警。

PostgreSQL CDCDebeziumKafka修改时间：2026-09-20 04:06:46