PostgreSQL逻辑复制解析WAL同步Kafka实战-平芜编程栈

简介：本资源面向数据库开发与数据集成工程师，提供一套基于PostgreSQL逻辑复制功能的实时数据变更捕获与同步系统源码。系统通过解析WAL日志捕获数据变更，将其转换为可执行的SQL语句，并借助Kafka消息队列实现PostgreSQL到异构数据源的实时同步，适用于大数据分析、实时报表与数据仓库等场景。压缩包共24个文件，约474KB，以17个Java源文件为核心实现，辅以2个XML配置、properties参数文件、README与说明文档及架构图，结构清晰便于二次开发。已有57人学习下载。读者可获得完整的CDC同步方案实现，涵盖WAL解析、SQL转换、Kafka发布订阅、事务回滚与故障恢复等关键模块，并附有配置说明与文档，便于快速理解逻辑复制原理并落地到实际项目中。

1. 从 WAL 到 Kafka：一条被低估的实时同步链路

PostgreSQL 的逻辑复制功能，本质上是在 WAL 日志层面开了一扇窗。很多团队第一次接触这个方向，是因为业务方提了一个很朴素的需求：订单库的变更要实时同步到 Kafka，下游做实时风控和数仓入湖。找了一圈方案，发现 Debezium 这类工具底层也是解析 WAL，于是开始琢磨能不能自己控制这条链路。这个标题讲的就是这件事——用 PostgreSQL 逻辑复制协议解析 WAL，把行级变更还原成 SQL 或结构化消息，再投递到 Kafka 等异构数据源。它适合两类人：一是需要把 PG 变更实时推给下游、又不想被第三方工具黑匣子绑死的后端工程师；二是正在做数据同步中间件、想搞清楚逻辑复制底层怎么跑的基础设施开发者。读完你应该能自己搭出一条最小可用的同步链路，并且知道哪些参数一改就翻车。

2. 逻辑复制与 WAL 解析：先搞清楚数据从哪来

2.1 逻辑复制槽到底在复制什么

PostgreSQL 的物理复制传的是 WAL 字节流，备库收到的是一模一样的页面块。逻辑复制不一样，它通过pgoutput插件把 WAL 里的物理记录解码成逻辑变更——INSERT、UPDATE、DELETE 的行级事件。这个解码过程依赖一个关键前提：表必须设置REPLICA IDENTITY，否则 UPDATE 和 DELETE 事件里只有主键，拿不到旧值。

逻辑复制槽（replication slot）是这条链路的锚点。它记录消费者已经确认到哪个 LSN，PG 会保证不清理这个位置之后的 WAL。这意味着如果消费者挂了太久，pg_wal目录会一直涨，磁盘撑爆是迟早的事。我一般会在创建槽的时候就把max_slot_wal_keep_size设上，比如 10GB，超了就让槽失效，宁可断同步也不能让主库挂掉。

-- 创建逻辑复制槽，使用 pgoutput 插件 SELECT * FROM pg_create_logical_replication_slot('sync_slot', 'pgoutput'); -- 查看槽的确认位点和保留的 WAL 量 SELECT slot_name, confirmed_flush_lsn, pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal FROM pg_replication_slots; -- 设置 WAL 保留上限，防止磁盘被撑爆 ALTER SYSTEM SET max_slot_wal_keep_size = '10GB'; SELECT pg_reload_conf();

第一段创建槽，pgoutput是 PG 10 以后内置的逻辑解码插件，不需要额外编译。第二段查retained_wal，这个值持续增长就说明消费者跟不上了。第三段是后悔药，设了上限之后槽最多保留 10GB WAL，超了槽会变成 invalid 状态，同步中断但主库安全。

2.2 发布订阅模型与 pgoutput 协议

PG 的逻辑复制用「发布-订阅」模型。发布端定义哪些表的哪些操作要发出去，订阅端（或者你自己写的消费者）通过复制协议拉取。自己实现消费者的话，走的是START_REPLICATION SLOT ... LOGICAL命令，服务端返回的是一串二进制消息，格式由pgoutput决定。

pgoutput 的消息类型主要有几种：B是 Begin，C是 Commit，R是 Relation（表结构元数据），I是 Insert，U是 Update，D是 Delete。每条 DML 消息里带着元组数据，文本格式还是二进制格式取决于发布端publish_via_partition_root和列类型。我一般让发布端用文本格式，解析起来省事，虽然体积大一点。

-- 创建发布，指定要同步的表 CREATE PUBLICATION sync_pub FOR TABLE orders, order_items; -- 查看发布状态 SELECT * FROM pg_publication_tables WHERE pubname = 'sync_pub'; -- 在订阅端（或消费者）发起复制 -- 实际代码里通过复制协议发送： -- START_REPLICATION SLOT sync_slot LOGICAL 0/0 ( -- proto_version '1', -- publication_names 'sync_pub' -- );

发布定义好之后，消费者连接上来发START_REPLICATION，PG 就从槽的确认位点开始推 WAL 解码后的消息。proto_version '1'是当前主流版本，支持流式传输大事务。如果你的 PG 是 14 以上，可以考虑proto_version '2'，支持两阶段提交的解码。

2.3 把 WAL 变更还原成 SQL 的映射逻辑

拿到 pgoutput 消息之后，核心工作是把I/U/D消息还原成可执行的 SQL 或者结构化 JSON。Insert 最简单，R消息给了列名和类型，I消息给了新值，拼一个 INSERT 就行。Update 麻烦一点，U消息里可能只有新值（如果REPLICA IDENTITY DEFAULT），旧值只有主键。Delete 同理，只有主键能拿到。

# 简化版 pgoutput 消息解析：把 Insert 消息转成 SQL def parse_insert(relation, tuple_data): """ relation: {'columns': [{'name': 'id', 'type': 'int4'}, ...]} tuple_data: {'columns': [{'value': '1'}, {'value': 'alice'}]} """ cols = [c['name'] for c in relation['columns']] vals = [] for col in tuple_data['columns']: v = col.get('value') if v is None: vals.append('NULL') elif col.get('kind') == 'text': vals.append("'" + v.replace("'", "''") + "'") else: vals.append(v) sql = f"INSERT INTO {relation['name']} ({','.join(cols)}) VALUES ({','.join(vals)})" return sql

这段代码只处理了文本格式的 Insert，实际生产里还要处理二进制格式、TOAST 列、数组类型。关键点是kind字段，pgoutput 会告诉你这个值是文本还是二进制。文本格式下字符串要转义单引号，二进制格式下得按类型反序列化。我一般会在 Relation 消息到达时缓存表结构，后续 DML 消息直接用缓存，避免每次查information_schema。

3. 从零搭一条 PG 到 Kafka 的同步链路

3.1 环境准备与 PG 参数配置

先确认 PG 版本，逻辑复制在 10 以上才内置，推荐 14 或 16，pgoutput的稳定性和功能都更好。postgresql.conf里必须改几个参数，改完重启生效。

# postgresql.conf 关键配置 wal_level = logical # 必须，否则没有逻辑解码信息 max_replication_slots = 10 # 按消费者数量设，留余量 max_wal_senders = 10 # 复制连接数上限 max_slot_wal_keep_size = '10GB' # 防止 WAL 撑爆磁盘

wal_level改成logical需要重启，这个没得商量。max_replication_slots和max_wal_senders可以动态改，但建议直接写进配置文件。改完用SHOW wal_level;确认，返回logical才算数。

Kafka 这边，如果只是本地验证，单节点就够。生产环境至少三个 broker，min.insync.replicas=2，acks=all。Topic 的 partition 数按下游消费能力定，我一般按 PG 表的写入 QPS 来估，单 partition 撑几千 TPS 没问题。

# 创建同步用的 topic，6 个分区，副本因子 3 kafka-topics.sh --create \ --topic pg-cdc-orders \ --partitions 6 \ --replication-factor 3 \ --bootstrap-server localhost:9092 # 确认 topic 状态 kafka-topics.sh --describe --topic pg-cdc-orders --bootstrap-server localhost:9092

分区数一旦定了不好改，改分区会打乱 key 的哈希分布。如果按主键哈希发消息，分区数变了同一个主键可能落到不同分区，下游按 key 聚合的逻辑就乱了。所以一开始宁可多分几个。

3.2 消费者代码：从复制流到 Kafka Producer

消费者要做三件事：连 PG 拉复制流、解析 pgoutput 消息、发到 Kafka。Python 里可以用psycopg2的复制协议支持，或者用asyncpg自己拼协议包。下面用psycopg2演示核心逻辑。

import psycopg2 from psycopg2.extras import LogicalReplicationConnection from kafka import KafkaProducer import json # 连接 PG，使用逻辑复制连接 conn = psycopg2.connect( "dbname=test user=repl_user host=pg-host port=5432", connection_factory=LogicalReplicationConnection ) cur = conn.cursor() # 创建复制槽（如果已存在会报错，捕获即可） try: cur.create_replication_slot('sync_slot', output_plugin='pgoutput') except psycopg2.errors.DuplicateObject: conn.rollback() # Kafka producer producer = KafkaProducer( bootstrap_servers=['kafka1:9092', 'kafka2:9092'], value_serializer=lambda v: json.dumps(v).encode('utf-8'), acks='all', retries=3 ) # 开始消费复制流 cur.start_replication( slot_name='sync_slot', options={'proto_version': '1', 'publication_names': 'sync_pub'}, decode=False ) def consume(msg): # msg.payload 是 pgoutput 二进制消息，需要按协议解析 event = parse_pgoutput(msg.payload) if event: producer.send('pg-cdc-orders', key=event['pk'].encode(), value=event) msg.cursor.send_feedback(flush_lsn=msg.data_start) cur.consume_stream(consume)

start_replication的decode=False表示拿原始字节，自己解析。consume_stream是阻塞循环，每收到一条消息调一次consume。关键在send_feedback，它告诉 PG 这个 LSN 之前的 WAL 可以释放了。如果忘了发 feedback，槽的confirmed_flush_lsn不动，WAL 会一直堆。

parse_pgoutput需要自己实现，按前面说的消息类型逐个解析。生产环境建议用现成的解析库，比如pgoutput的 Python 实现，或者直接用 Debezium 的解析逻辑但自己控制投递。自己从头写解析器容易在 TOAST 列和类型边界上翻车。

3.3 消息格式设计与下游消费约定

发到 Kafka 的消息格式决定了下游怎么用。我一般用 JSON，字段包括op（操作类型）、table、before（旧值）、after（新值）、ts（时间戳）、lsn。before只在 Update 和 Delete 时有值，而且只有主键列，这是REPLICA IDENTITY DEFAULT的限制。

{ "op": "U", "table": "orders", "before": {"id": 1001}, "after": {"id": 1001, "status": "paid", "amount": 299.00}, "ts": "2025-01-15T10:30:00Z", "lsn": "0/1A2B3C4D" }

下游消费的时候，op决定怎么处理：I做 upsert，U做 upsert，D做 delete。lsn可以用来做幂等，同一个 LSN 的消息重复消费直接丢弃。ts是 PG 事务提交时间，不是消息发送时间，做时间窗口聚合的时候用这个。

提示：如果下游需要完整的旧值，把表的REPLICA IDENTITY改成FULL，这样 Update 和 Delete 消息里会带所有列的旧值。代价是 WAL 体积变大，写入性能会降。

4. 避坑与排查：那些让同步链路断掉的细节

4.1 复制槽不释放导致磁盘写满

现象：主库磁盘使用率持续上涨，pg_wal目录越来越大，查pg_replication_slots发现retained_wal几十 GB。

原因：消费者挂了或者消费太慢，confirmed_flush_lsn不推进，PG 不敢清理 WAL。如果没设max_slot_wal_keep_size，WAL 会一直堆到磁盘满。

解决：先设max_slot_wal_keep_size，给个硬上限。然后查消费者为什么慢——是解析逻辑太重、Kafka 发送阻塞、还是网络问题。如果是消费者彻底挂了，要么修好继续消费，要么删槽重建，但删槽意味着中间的数据变更丢了，得先确认下游能不能接受。

4.2 REPLICA IDENTITY 没设导致 Update 丢旧值

现象：下游收到的 Update 消息里before只有主键，没有其他列的旧值，做增量对比的时候拿不到变更前的状态。

原因：表的REPLICA IDENTITY默认是DEFAULT，只记录主键。逻辑复制要拿到完整旧值，必须改成FULL。

解决：ALTER TABLE orders REPLICA IDENTITY FULL;。改完之后新产生的 WAL 会带完整旧值，但已有的槽不会回溯，得等新事务。注意FULL会让 WAL 体积明显增大，写入密集的表要评估磁盘和复制延迟。

4.3 大事务导致消费者内存溢出

现象：消费者进程 OOM 被杀，日志里看到某个事务的变更消息特别大。

原因：pgoutput 在proto_version '1'下，大事务的所有变更会在 Commit 时一次性推给消费者。如果一个事务改了十万行，消费者要一次性收十万条消息再处理，内存扛不住。

解决：升级到proto_version '2'，支持流式传输大事务，变更消息在事务进行中就逐步推送。或者业务侧拆大事务，单事务变更控制在几千行以内。再不行就在消费者里做流式处理，收到一条处理一条，不要攒着。

4.4 Kafka 发送阻塞拖垮整个消费链路

现象：PG 复制延迟越来越大，查消费者发现卡在 Kafkasend调用上。

原因：Kafka broker 响应慢或者网络抖动，producer.send是异步的但缓冲区满了会阻塞。如果消费者是单线程，Kafka 一卡，PG 复制流就不消费了，feedback 不发，WAL 开始堆。

解决：给 producer 设max_block_ms，比如 5000ms，超了就抛异常而不是无限等。消费者里做批量发送，攒一批再发，减少网络往返。再不行就加本地磁盘队列做缓冲，Kafka 恢复后补发。但本地队列要考虑重启丢消息的问题，得配合 LSN 做断点续传。

4.5 表结构变更导致解析器崩溃

现象：PG 上执行了ALTER TABLE ADD COLUMN，消费者解析后续消息时报错退出。

原因：pgoutput 的 Relation 消息会在表结构变更后重新发送，解析器如果没处理新列，元组解析就会错位。

解决：解析器要动态处理 Relation 消息，每次收到就更新本地缓存的表结构。不要硬编码列顺序和数量。另外 DDL 变更期间最好暂停同步，等结构稳定了再恢复，避免中间态消息解析出错。

5. 进阶：用 LSN 做断点续传与幂等消费

同步链路最怕的是消费者重启后不知道从哪继续。逻辑复制槽本身记录了confirmed_flush_lsn，重启后从槽的位点继续拉就行。但如果你在消费者里做了异步发送，Kafka 的确认和 PG 的 feedback 不同步，重启后可能重复消费或者丢消息。

我的做法是：Kafka 发送成功后再发 PG feedback，保证「至少一次」。下游用 LSN 做幂等，同一个 LSN 的消息重复到达直接丢弃。LSN 在 pgoutput 的 Begin 和 Commit 消息里都有，解析出来存到消息体里。

# 幂等消费示例：用 Redis 记录已处理的 LSN import redis r = redis.Redis(host='redis-host', port=6379) def process_message(event): lsn = event['lsn'] # SETNX 原子操作，key 不存在才设置成功 if not r.setnx(f"cdc:lsn:{lsn}", "1"): return # 已处理过，跳过 r.expire(f"cdc:lsn:{lsn}", 86400) # 24 小时过期 # 实际业务处理 do_sync(event)

setnx保证同一个 LSN 只处理一次，expire防止 Redis 无限增长。24 小时过期意味着如果消费者停了超过 24 小时再重启，可能重复处理之前的消息，所以过期时间要大于最大可能的重启间隔。

验证同步是否完整，可以定期对比源表和目标表的行数和校验和。PG 侧用pg_current_wal_lsn()拿当前位点，等消费者追到这个位点后，在目标端做 count 和 checksum 对比。不一致就说明中间有丢消息，得查 feedback 逻辑和 Kafka 的 acks 配置。