上周又被问了一次那个老问题：业务库新加了一张订单表，报表侧要把 T+1 的离线跑批改成准实时，怎么搞。我第一反应还是那套组合拳——用 Debezium 抓 PostgreSQL 的逻辑复制流，再用 RocketMQ Connect 把变更事件接进消息队列分发到下游。这套 PostgreSQL 实时同步的方案我从第一次上线到现在，踩过的坑能写满两页 A4 纸：复制槽把磁盘撑爆、快照把主库锁了几分钟、numeric 字段精度莫名其妙少两位、下游消费端重复写导致对账天天吵架。

这篇就把整套东西从头到尾捋一遍。Debezium负责“读得准”，它伪装成一个 PostgreSQL 的从库，靠逻辑复制协议把每一行数据的 insert/update/delete 原样读出来；RocketMQ Connect负责“送得到”，它把 Debezium 产出的变更事件当成 source 数据，写进 RocketMQ 的 Topic，再由 sink 连接器落到目标端。两者拼起来，就是一条不侵入业务代码、不靠定时扫表、延迟能压到秒级的同步链路。

适合谁看：手上有一张或几张 PostgreSQL 表要实时同步到别的库、缓存、搜索或者数仓，又不想改业务代码的同学；被 DataX、Sqoop 那种定时全量+增量折磨过，想换 CDC 思路的同学；以及已经在用 RocketMQ，想复用它做数据管道的同学。下面按我实际落地的顺序讲，配置、参数、算账、排查都在里面，能直接抄。

1. 方案整体设计与选型思路拆解

动手之前先把图纸画清楚，不然装到一半发现方向错了，返工成本比想象高得多。

1.1 为什么选 CDC 而不是定时拉取

定时拉取的本质是“猜”。你设一个 5 分钟窗口，用update_time > 上次时间去捞增量，看起来能用，实际有三个绕不过去的硬伤。第一个是漏数据：同一秒内先插入再更新，两次操作的update_time可能落在同一个时间戳上，下一次窗口按>比较就把中间那条状态跳过去了；如果业务用的是数据库服务器时间而不是应用时间，主从切换时时钟回拨还会再漏一批。第二个是删不掉：物理删除在源表里连痕迹都没有，你永远不知道哪一行没了，只能靠定期全量比对，成本高得离谱。第三个是压主库：每次扫描都是实打实的索引范围扫描，表大了之后update_time索引的碎片会让你怀疑人生。

CDC 走的是另一条路。PostgreSQL 的逻辑复制会把每一行变更以逻辑日志的形式吐出来，包含操作类型、变更前后的完整字段值、事务提交时间、LSN 位点。这些信息是数据库自己产生的，不需要你去猜。删除就是op = 'd'，更新就是op = 'u'且带着before和after两份快照，你拿到的东西和数据库内部认知完全一致。延迟也不再取决于你设的窗口大小，而是链路上每一跳的处理耗时，正常情况下一两百毫秒就能到下游。

我做过一次对比：一张 800 万行的订单表，用定时拉取方案做 1 分钟窗口，高峰期主库 CPU 会多出 8 到 12 个百分点；换成 CDC 之后，Debezium 这个“假从库”常驻占用不到 1 个核，WAL 解析的额外开销在主库侧基本测不出来。当然 CDC 不是免费的，它换来的是对 WAL 保留策略的强依赖，这个后面单独讲。

1.2 Debezium 与 RocketMQ Connect 各自负责什么

很多人第一次接触会 confused：Debezium 不是自带 Kafka Connect 吗，为什么还要套一层 RocketMQ Connect？

这里要分清两个概念。Debezium 的核心是一个数据库连接器（connector），它实现了“从数据库读变更、转成结构化事件”这件事，本身不关心事件往哪送；它跑在 Kafka Connect 这个运行时框架里，是因为历史上它和 Kafka 生态绑定最深。而RocketMQ Connect 是另一套连接器运行时，它做的事情和 Kafka Connect 高度类似：管理连接器的生命周期、分配 task、维护 offset、提供 source/sink 两类插件接口。

所以在这套方案里，分工是这样的：Debezium 提供 PostgreSQL 的读取能力（逻辑复制协议解析、快照、schema 变更感知），RocketMQ Connect 提供运行时的调度、容错和消息投递能力，把 Debezium 产出的事件写进 RocketMQ 的 Topic。选 RocketMQ Connect 而不是 Kafka Connect 的理由很实在：如果你的消息中间件本来就是 RocketMQ，多引入一套 Kafka 只为了跑 Connect，运维要维护两套集群、两套监控、两套权限体系，纯属给自己找事。RocketMQ Connect 的 source 连接器能直接复用现有 NameServer 和 Broker，运维面收敛很多。

需要说明的是，RocketMQ Connect 的 Debezium 相关连接器在社区里的成熟度和 Kafka Connect 版本有差距，部分 Kafka Connect 的 SMT（单消息转换）插件不能直接搬过来用。这一点在方案设计阶段就要认，别指望配置照抄。我在实际项目里就把字段裁剪和格式转换挪到了下游消费端，连接器只做“原样搬运”，链路反而更稳。

1.3 组件版本与部署拓扑

版本这块吃过亏，必须写死。我当前验证过比较稳的一组：PostgreSQL 13 及以上（逻辑复制的pgoutput插件在 10 就有了，但 13 之后发布/订阅的行为更完善），Debezium 连接器 2.x 系列，RocketMQ 4.9.x 及以上，RocketMQ Connect Runtime 与连接器版本对齐。

部署拓扑我推荐三种，按规模选：

绝大多数团队从第二种起步就够了。不要一上来就搞集群，RocketMQ Connect 的 task 分配在节点扩缩容时会发生重平衡，重平衡期间 source 连接器可能重新拉快照或者短暂停滞，这个抖动在业务低峰期做比较安全。等单节点 CPU 稳定超过 70% 或者出现明显的处理积压，再考虑加节点。

还有一点容易被忽略：Connect Runtime 不要和 Broker 混部。Broker 是 IO 密集型，负责消息的读写和刷盘；Connect 是 CPU 和内存密集型，要做反序列化、格式转换、批量组装。混在一起跑，一次大的快照就能把 Broker 的页缓存冲掉，消息发送延迟毛刺会非常明显。我用过的最小规格是 4 核 8G 单独跑 Runtime，配合 8 核的 Broker 集群，这套配置撑住过日均两亿条事件。

2. PostgreSQL 侧的准备与核心配置细节

源头没配置好，后面全是白费功夫。这一章全部是数据库侧的事，改完需要重启的参数我会标出来。

2.1 逻辑复制的前置条件与必须改的参数

PostgreSQL 的逻辑复制不是开箱即用的，要在postgresql.conf里改几个参数：

# 必须改成 logical，否则连接器连上来会直接报错 wal_level = logical # 至少要比并发连接器数量大，每个复制槽占一个 sender max_wal_senders = 10 # 复制槽数量上限，留够余量，因为下线连接器后槽不会自动消失 max_replication_slots = 10 # 单个事务解码时的内存上限，大事务容易在这里出问题 logical_decoding_work_mem = 256MB

wal_level改成logical需要重启，这是硬要求，找个维护窗口做。改动之后 WAL 的写放大比例会上升，因为要额外记录足够还原行级变更的信息，磁盘容量要预估充分，我一般按原来的 1.3 到 1.5 倍留空间。

账号权限这块，很多人卡在这里。Debezium 用的账号需要两个能力：登录并具备REPLICATION属性，以及对要同步的表有SELECT权限。注意不是超级用户也能做，最小化授权是这样的：

CREATE ROLE cdc_user WITH LOGIN REPLICATION PASSWORD '你的密码'; GRANT CONNECT ON DATABASE your_db TO cdc_user; GRANT USAGE ON SCHEMA public TO cdc_user; GRANT SELECT ON ALL TABLES IN SCHEMA public TO cdc_user; -- 后续新建的表也要自动带上权限，这行不能少 ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO cdc_user;

最后那句ALTER DEFAULT PRIVILEGES是血泪教训。当时业务新加了一张表，连接器配置里table.include.list是通配的，结果新表一上线就报权限不足，连接器直接挂了，重启之后还因为 offset 问题重新拉了一次全量快照。只要有通配同步的打算，这行就必须加。

注意：REPLICATION属性和SUPERUSER是两回事，不要图省事直接给超级用户。CDC 账号在链路上是只读的，权限越收敛，出问题时排查范围越小。

2.2 复制槽与 WAL 保留，这条命脉必须盯住

逻辑复制槽（replication slot）是这套方案里最危险的东西，没有之一。

它的工作机制是：数据库记住“这个消费者已经确认到哪个 LSN 了”，在此之前的 WAL 文件即使 checkpoint 也不能删，因为消费者可能还要来读。如果消费者停止消费（连接器挂了、网络不通、账号被锁），数据库会一直保留 WAL，直到把磁盘写满。磁盘写满之后 PostgreSQL 会拒绝所有写请求，整个业务库躺平。

我遇到过一次：测试环境把连接器停了去调参数，忘了复制槽还在，两天后磁盘告警，pg_wal目录涨到了 200 多 G。所以生产环境必须做三件事。

第一件，给复制槽配监控。查当前所有槽的滞留情况：

SELECT slot_name, active, pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal, pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) AS unflushed FROM pg_replication_slots;

retained_wal就是还没被回收的 WAL 量，超过阈值（我设的是 20G）就告警。active = false且retained_wal持续增长，说明消费者已经掉了，赶紧去查连接器。

第二件，给 WAL 目录做容量兜底。max_wal_size控制的是检查点触发的阈值，控制不了复制槽滞留，真正的保险是文件系统的磁盘配额告警加上定时巡检。

第三件，定期清理废弃的复制槽。连接器下线时如果没走正常流程，槽会残留。删除前一定要确认没有活跃消费者：

-- 先确认 active 是 false SELECT slot_name, active FROM pg_replication_slots WHERE slot_name = '你要删的槽名'; -- 确认后再删，这条命令会造成对应消费者无法继续，务必谨慎 SELECT pg_drop_replication_slot('你要删的槽名');

如果pg_drop_replication_slot报“复制槽正在使用”，说明还有进程连着，先用pg_terminate_backend把对应的 walsender 干掉再删。别在业务高峰期做这个操作，因为强制断开会导致连接器重连并可能触发重新快照。

2.3 表结构、主键与那些容易翻车的字段类型

Debezium 对表结构有几个隐含要求，踩过就知道疼。

主键必须有。没有主键的表，Debezium 在快照阶段会报错退出，因为它没法给每一行生成稳定的唯一标识。有些人会想“我加个唯一索引行不行”，答案是不行，必须是主键或者被显式配置为标识列的字段。如果实在没有主键，可以在连接器里用message.key.columns指定一组列作为逻辑主键，但要保证这组列的组合唯一，否则下游会出现覆盖写。

UPDATE 和 DELETE 的旧值问题。PostgreSQL 默认的REPLICA IDENTITY是DEFAULT，只把主键的旧值放进日志里。这意味着 Debezium 拿到的before字段只有主键，其他列全是 null。绝大多数场景这没问题，因为你按主键做 upsert 就够了。但如果下游需要“根据某个字段的旧值做判断”（比如状态机流转、变更前后对比），就要改成FULL：

ALTER TABLE your_table REPLICA IDENTITY FULL;

代价是 WAL 量显著增长，因为每一行变更都要带上整行的旧值。只在确有必要的那几张表上开，别全库开，我见过一整库开 FULL 之后 WAL 写入量翻三倍的情况。

字段类型这块是重灾区。numeric和decimal默认会被 Debezium 转成二进制格式（Kafka Connect 的Decimal逻辑类型），如果下游序列化没处理好，就会出现“明明入库是 12.34，下游收到 1234”这种精度错位。解决办法是在连接器里显式指定：

decimal.handling.mode=string

改成string之后，数值会以字符串形式传输，下游自己决定用什么精度解析，我个人强烈推荐这个配置，因为金融类数据一旦精度出错，排查成本极高。

时间类型也坑。timestamp without time zone在 Debezium 事件里会被当作 UTC 处理，如果业务实际存的是本地时间，下游就会莫名其妙差 8 小时。处理方式有两种：要么统一在数据库层用timestamptz，要么把time.precision.mode和时区转换逻辑在消费端显式处理。我在项目里定了个规矩：新表一律用timestamptz，老表在连接器里加一层字段转换，别指望它自动帮你猜。

另外，bytea和text这类大字段在大表里会被 PostgreSQL 存到 TOAST 表里。当一行数据很大、且更新没触及 TOAST 列时，日志里不会带上这部分列的实际值。表现就是下游偶尔收到这个字段是 null，但实际上源库有值。规避方式是对这类字段做单独的查询补齐，或者干脆不把它们放进同步范围——别骗自己“应该不会出问题”。

3. Debezium 采集端的搭建与关键参数

这一章是整套链路的核心，参数多，我一个一个拆。

3.1 部署方式与连接器配置骨架

RocketMQ Connect 的 source 连接器是以 JSON 配置提交的，提交方式是通过 Runtime 的 REST 接口 PUT 一个配置文件。配置的骨架长这样：

{ "connector-class": "org.apache.rocketmq.connect.debezium.postgresql.DebeziumPostgresConnector", "topic": "cdc-order-topic", "namesrvAddr": "你的NameServer地址:9876", "srcClusterName": "DefaultCluster", "connectorConfig": { "connector.class": "io.debezium.connector.postgresql.PostgresConnector", "database.hostname": "源库地址", "database.port": "5432", "database.user": "cdc_user", "database.password": "你的密码", "database.dbname": "your_db", "topic.prefix": "pg_prod", "schema.include.list": "public", "table.include.list": "public.order,public.order_item", "plugin.name": "pgoutput", "slot.name": "debezium_order_slot", "publication.autocreate.mode": "filtered", "snapshot.mode": "initial", "decimal.handling.mode": "string", "heartbeat.interval.ms": "10000", "tombstones.on.delete": "true", "max.batch.size": "2048", "max.queue.size": "8192", "poll.interval.ms": "500" } }

提交的时候注意 JSON 层级，connector-class和topic是 RocketMQ Connect 这一层要认的字段，connectorConfig里面才是 Debezium 原生的参数。这两层混了是最常见的启动失败原因，我第一次配的时候把slot.name写到外层，连接器起来了但就是读不到数据，日志里只有一句语焉不详的警告，找了半天。

topic这个字段要注意，它是 RocketMQ 侧的 Topic 名称。Debezium 自己会按topic.prefix + 表名生成内部主题标识，最终在 RocketMQ 里落到哪个队列，取决于连接器的映射逻辑。多表同步的场景，我建议按表拆分多个连接器实例，每个实例一个 Topic，消费端按业务归属分别订阅，隔离性更好，出问题也不会互相牵连。

3.2 snapshot.mode 选错，等于开局就输

snapshot.mode是我认为最需要想清楚的参数，它决定了连接器启动时怎么处理“存量数据”。

initial是默认值，第一次启动时先做一次全表快照，把存量数据当成一批 insert 事件发出去，然后接上增量。这个模式适合全新的同步链路。但要注意，快照期间 PostgreSQL 会加锁。pgoutput插件在快照开始时需要获取表的ACCESS SHARE锁，并短暂持有，大表上这个锁虽然不阻塞读写，但会阻塞 DDL。如果你在快照期间对这张表做了ALTER TABLE，两边会互相等待直到超时。

never是纯增量，启动后从当前 LSN 开始读，不做存量快照。适用于“存量已经通过别的通道灌好了，只需要接增量”的场景。

schema_only只快照表结构不导数据，后面接增量。

initial_only只做一次快照就停，适合做离线初始化。

我的经验是分阶段做：先用initial_only在业务低峰期把存量灌完，把下游的基线数据建好；再把配置切成never重启，接纯增量。这样有两个好处：快照阶段的锁和压力被隔离在一个可控窗口里；增量阶段重启不会触发重新快照，风险小得多。

如果坚持一步到位用initial，务必算清楚快照要跑多久。快照的速率大致取决于单行大小、网络带宽和下游写入能力，我实测过一个中等规格的实例，每秒能吐 3 到 5 万行。一张 5000 万行的表大概要跑 20 到 30 分钟，这期间 WAL 会持续累积（因为快照期间的变更也要保留给增量阶段消费），磁盘余量必须按快照时长乘上写入速率预留。

3.3 心跳、批量与几个保命参数

heartbeat.interval.ms这个参数看着不起眼，但它是防 WAL 堆积的关键。如果一张表长时间没有变更，Debezium 就不会产生任何位点更新，复制槽的confirmed_flush_lsn会停在原地不动，WAL 就一直不释放。心跳机制的作用是：即使没有数据变更，也定期往复制流里打一个空事件，推动位点前进。

我一般设 10 秒。设得太小会增加无效流量，设得太大（比如 5 分钟）在低峰期还是会积压。这个参数配合 2.2 节的监控一起用，效果最好。

max.batch.size和max.queue.size控制的是内部缓冲。前者是每次从数据库拉取的变更条数，后者是排队等待发送的队列长度。默认值偏保守，在高峰期容易成为瓶颈，我的经验值是max.batch.size设 2048，max.queue.size设 8192。但要注意，队列越大，连接器崩溃时丢在队列里未确认的数据越多，重启后会重放，重复量也越大。如果下游对重复敏感，队列反而要设小一点。

poll.interval.ms是拉取间隔，设 500 毫秒比较平衡。设成 10 毫秒会把 CPU 打满（空转），设成 5 秒延迟就上去了。

tombstones.on.delete这个参数控制删除时是否发送一条 key 为 null 的墓碑消息。在 Kafka 生态里它是用来让日志压缩清理旧 key 的；在 RocketMQ 场景下这个机制不完全对应，我建议设成 false 或者在下游直接忽略，否则消费端每次删除都会收到两条消息（一条 delete，一条空值），一不小心就把数据写成空。

还有一个容易被忽略的：publication.autocreate.mode。设成filtered表示连接器会自动创建一个只包含table.include.list中表的 publication，这是最省事的方式。但如果你的账号没有建 publication 的权限，就得手动建好再设成disabled：

CREATE PUBLICATION cdc_pub FOR TABLE public.order, public.order_item;

手动建的好处是可控，坏处是每次加表都要改 publication。团队协作的场合我更倾向于手动建，因为变更动作可见、有记录。

4. RocketMQ Connect 侧的落地实现

采集端跑通之后，事件能不能稳稳落到 Topic，就看这一层了。

4.1 Runtime 准备与 offset 存储配置

RocketMQ Connect 的 Runtime 是一个独立进程，核心配置在connect.conf里：

# 节点标识，集群内必须唯一 workerId=cdc-worker-01 # 配置和 offset 的存储位置，RocketMQ 用 Topic 存 storePathRootDir=/data/rocketmq-connect/store namesrvAddr=你的NameServer地址:9876 httpPort=8082 # 内部使用的 Topic，首次启动会自动创建 configTopicName=connect-config offsetTopicName=connect-offset

这里的关键是offset 存储。source 连接器读到哪个位点，是靠这个存储来记的。单节点用本地文件配置也能跑，但只要涉及重启恢复，就必须用可靠存储。我强烈建议用 RocketMQ Topic 作为 offset 存储，因为它是持久化的、有副本的，节点挂了不丢位点。

有个细节值得说：offset 提交是有间隔的，不是每条消息都提交。这意味着连接器异常退出时，会有一小段数据被重复消费（从上次提交的位点重放）。这是 at-least-once 语义的必然结果，不是 bug。下游必须按主键做幂等写入，这个后面 4.3 节细讲。

Runtime 启动之后，用 REST 接口管理连接器：

# 提交或更新连接器 curl -X PUT -H "Content-Type: application/json" \ -d @order-connector.json \ http://localhost:8082/connectors/cdc-order # 查看状态 curl http://localhost:8082/connectors/cdc-order/status # 停止并删除 curl -X DELETE http://localhost:8082/connectors/cdc-order

status返回里会包含每个 task 的状态和失败原因，这是排查的第一入口。如果状态一直是RUNNING但消费端没数据，问题就不在连接器本身，往下游链路查。

4.2 消息格式映射与下游 sink 的对接

Debezium 吐出来的原始事件结构是这样的（简化版）：

{ "before": null, "after": { "id": 1001, "order_no": "SO20240115001", "amount": "128.50", "status": "PAID", "created_at": "2024-01-15T10:23:45Z" }, "source": { "db": "your_db", "schema": "public", "table": "order", "lsn": 123456789 }, "op": "c", "ts_ms": 1705314225000 }

op字段是操作的语义：c是插入，u是更新，d是删除，r是快照读。这个字段是整个链路的分发依据。

下游 sink 连接器我一般配两类目标。第一类是另一个关系库（比如目标 PostgreSQL 或者分析型库），写入逻辑是：c和u走 upsert，d走按主键删除。第二类是消息分发型，直接把事件转发给业务方订阅，由业务自己决定怎么处理。

落库的 upsert 语句长这样，以 PostgreSQL 目标端为例：

INSERT INTO order_target (id, order_no, amount, status, created_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT (id) DO UPDATE SET order_no = EXCLUDED.order_no, amount = EXCLUDED.amount, status = EXCLUDED.status, created_at = EXCLUDED.created_at;

ON CONFLICT是幂等的关键。如果你用的是 MySQL 目标端，对应的是INSERT ... ON DUPLICATE KEY UPDATE，逻辑一样。千万别用“先 delete 再 insert”这种写法，一方面性能差一倍以上，另一方面在并发消费同一主键时会出现短暂的“数据不存在”窗口，读者会读到脏状态。

字段映射这块有个坑：Debezium 的时间戳是毫秒数或者 ISO 字符串，目标库字段如果是timestamp，要在写入前显式转换，别指望驱动自动转。金额字段如果在连接器里设了decimal.handling.mode=string，写入前要转回numeric，否则插入会报类型不匹配。

4.3 顺序性、幂等与事务边界

这三个词是 CDC 链路上最有争议的部分，我讲我的实际做法。

顺序性。RocketMQ 保证的是队列内有序，不保证 Topic 全局有序。要做到“同一行数据的变更按顺序被消费”，必须让同一主键的消息落到同一个队列。做法是在发送时用主键作为分区键（sharding key），RocketMQ 会按这个 key 的哈希选队列。这里有个前提：连接器必须正确地把主键提取出来当作消息 key。Debezium 的变更事件里主键信息在key部分，RocketMQ Connect 的 source 框架会用它做 key。如果表没有主键或者主键提取失败，消息会随机分布，顺序性就没了，表现是下游偶尔出现“更新在插入之前到达”的报错。

幂等。前面说的 at-least-once 决定了下游一定会收到重复消息。解法就是在写入侧做幂等，ON CONFLICT已经覆盖了插入和更新。删除操作要小心：如果一条删除消息重放两次，第二次执行时行已经不存在，DELETE影响 0 行，这没问题；但如果删除之后源库又插入了一条相同主键的记录，而重复的旧删除消息在插入之后到达，就会把新数据删掉。这个场景在“主键复用”的业务里是真实存在的。我的处理方式是给删除操作加版本号或时间戳判断，只有当前消息的ts_ms大于目标行记录的更新时间，才允许删除。

事务边界。一个事务里改了 100 行，Debezium 会拆成 100 条独立事件发出去，每条都带同一个事务标识（source.txId或类似字段）。如果下游要求“要么全成功要么全失败”，就必须在消费端按事务标识做聚合，收齐之后再统一提交。这在 RocketMQ 里要靠自定义消费逻辑实现，成本不低。

我的建议是先评估业务是否真的需要跨行事务。订单主表和明细表在一个事务里写入，但下游如果只是做报表查询，短暂的不一致（几十毫秒）完全可以接受，那就别做聚合，简单可靠得多。如果确实是资金类业务必须强一致，那就在消费端按txId攒批，设置一个超时窗口（比如 2 秒），超时后强制提交，避免因为个别事件丢失导致永远凑不齐。

5. 端到端联调与数据一致性校验

链路搭起来不等于数据对得上，验证这一步不能省。

5.1 全链路验证的标准动作

我每次上线都走这个流程，五步，一步都不能跳过。

第一步，造一条最小变更。在源库插入一行，值全部用可识别的标记（比如order_no = 'TEST_CDC_001'），然后去 RocketMQ 控制台看对应 Topic 有没有消息。这一步只验证“通不通”，不验证“对不对”。

第二步，验证三种操作。插入、更新（改一个字段）、删除，各做一次，确认op字段分别是c、u、d，after和before内容符合预期。特别要检查before在更新场景下是否只有主键（这是REPLICA IDENTITY DEFAULT的正常表现，不是 bug）。

第三步，验证目标端落库。去目标库查那条测试数据，字段值、类型、时间是否和源库一致。重点核对numeric的精度和timestamp的时区，这两个是最容易出错的。

第四步，验证断点续传。把连接器停掉，期间在源库做几次变更，再启动连接器，看这几条变更是否被补上。这一步验证的是 offset 提交和复制槽的配合是否正常。

第五步，验证重复消费的容忍度。手动把连接器重启两次，观察下游是否有重复数据或者报错。如果下游没做幂等，这一步必然会暴露问题。

这五步做完，基本能覆盖 80% 的常见问题。

5.2 存量数据的对账方法

增量跑通之后，最大的不确定性在存量。快照阶段的数据和源库是不是完全一致？

小表可以直接比对行数和关键字段的聚合值：

-- 源库 SELECT count(*), sum(amount), max(created_at) FROM public.order WHERE created_at < '你的快照截止时间'; -- 目标库执行相同查询，两边数字必须完全一致

行数一致但内容不一致的情况也常见，这时要做抽样比对。我的做法是按主键区间分段，每段取 1000 行，用程序把两边的整行数据做哈希比对：

import hashlib def row_hash(row): # 字段顺序固定，null 用特殊标记，避免 None 和 'None' 混淆 parts = [] for col in sorted(row.keys()): val = row[col] parts.append(f"{col}={'__NULL__' if val is None else val}") return hashlib.md5("|".join(parts).encode()).hexdigest()

两边分别算出主键到哈希的映射，做集合差。差异行数超过万分之一就要停下来查原因，通常是对齐问题（时区、精度、字符编码）而不是真的丢了数据。

大表对账不要一次跑完，会拖垮源库。按主键范围切片，每片跑完 sleep 一会儿，控制在业务低峰期执行。我做过一次 3000 万行的对账，按 5 万一行切片，跑了差不多 40 分钟，源库负载基本无感。

5.3 性能调优的实测数据

不同规模下的参数差异很大，我把自己测过的几组贴出来供参考。测试条件是单表 2000 万行存量、日增量 500 万条事件、单条事件平均 800 字节。

激进配置看着诱人，但要注意“重启重放量”这个隐性成本。队列里堆着 3 万多条未确认消息，一旦进程崩溃，重启后要把这些全部重放一遍，下游要承受一波重复写入的冲击。在业务对重复特别敏感的场景，宁可延迟高一点也要把队列压小。

还有一个调优点在 PostgreSQL 侧：logical_decoding_work_mem设大之后，大事务的解码会走内存而不是溢写到磁盘，速度提升明显。我从 64MB 调到 256MB，一个 10 万行的批量更新事务处理耗时从 12 秒降到了 3 秒左右。但这个参数是全局的，设太大会挤占正常工作内存，建议不超过实例总内存的 10%。

6. 常见问题与排查技巧实录

这一章全是实战里撞出来的，按现象查表最快。

6.1 典型故障速查表

这张表我打印出来贴在工位上过，新同事上手时对着查，能省掉大量“盲猜”时间。

6.2 我踩过的三个印象最深的坑

第一个坑是复制槽的命名冲突。开发环境和测试环境共用了一套数据库的不同 schema，两边连接器配置里slot.name用了同一个默认值，结果第二个连接器启动时直接报“复制槽已被占用”。更麻烦的是，第一个连接器因为 publication 不匹配，什么数据都没读出来，但槽已经建好了，白白占着 WAL。教训是：每个连接器的 slot.name 必须带环境和业务标识，全局唯一。我现在统一用{环境}_{业务}_{表名}_slot这个格式，一眼能看出归属。

第二个坑是快照把主库 IO 打满。那次是给一张 1.2 亿行的表做初始化，配置里snapshot.fetch.size用了默认值，快照阶段发出大量小查询，把主库的随机 IO 拉满，业务侧的慢查询告警响了十几分钟。后来改成：

snapshot.fetch.size=10240 snapshot.max.threads=1

fetch.size调大之后，单次网络往返拿到的行数变多，查询次数下降一个数量级。max.threads保持 1 是刻意为之，多线程快照虽然快，但对主库的并发压力也成倍增加，在源库不能停服的场景下，慢就是快。

第三个坑最隐蔽：连接器重启后不消费新数据。现象是状态显示 RUNNING，但下游一条新数据都没有，日志里只有心跳在跳。查了半天发现是 offset 存储里的位点比复制槽的confirmed_flush_lsn还要新，连接器认为“我已经读到很后面了”，于是从那个位置开始等，但源库的位点还没到那里，两边就卡住了。原因是之前手动改过 publication 并重建了复制槽，槽的起点被重置了，而 offset 存储还留着旧值。处理方式是清掉这个连接器的 offset 记录重新提交，或者停掉连接器后删除复制槽让它重建。这个问题没有任何报错，纯靠对比两边的 LSN 才能发现，建议把“offset 位点”和“复制槽位点”的差值做成一个监控指标。

6.3 上线后必须配的监控项

链路跑起来只是开始，没有监控的 CDC 链路等于埋了个雷。这四项是底线。

第一项，复制槽滞留量。前面给过查询语句，超过 20G 告警。这一项能防住 90% 的严重事故。

第二项，连接器状态。通过 REST 接口定时拉取status，任何 task 不是 RUNNING 就告警。注意要区分“连接器重启中”和“真的挂了”，加个连续三次异常的判断，避免抖动误报。

第三项，端到端延迟。做法是在源库变更的事件里带ts_ms，消费端拿到之后用当前时间减去它，得到单条延迟。按分钟统计 P99，超过 5 秒告警。这个指标比“消息堆积量”更能反映真实体验。

第四项，目标端写入失败率。任何写入报错都要抓，特别是主键冲突、类型不匹配、字段超长这类。这类错误通常不会让连接器挂掉，但会导致数据静默丢失，比崩溃更可怕。建议把失败的消息原样落到一个死信 Topic 里，方便事后补数。

还有一个小技巧：给每条事件打个链路标识（比如在连接器配置里加一个自定义 header），从源库到目标端全链路可追踪。排查“某条数据为什么没同步过来”的时候，有这个标识能省掉大量日志翻找的时间。

最后说个我个人一直坚持的习惯：每次改连接器配置之前，先把当前配置导出备份。RocketMQ Connect 的配置存在 Topic 里，误删了不好恢复；更重要的是，改了参数之后如果效果不好，你得能快速回滚到之前那套跑得稳的配置。我吃过一次亏，半夜调参把snapshot.mode从never改成了initial，重启之后整表重新快照，主库负载直接拉满，那晚的教训是——所有涉及快照模式的改动，必须放在变更窗口里做，并且提前想清楚 WAL 和 IO 的余量。