PostgreSQL 逻辑复制与 CDC：发布订阅、WAL 解码、零停机迁移与 Debezium/Kafka 集成

-- 开启逻辑复制所需的配置
ALTER SYSTEM SET wal_level = 'logical';
ALTER SYSTEM SET max_replication_slots = 10;
ALTER SYSTEM SET max_wal_senders = 10;
SELECT pg_reload_conf();

-- 直接观察 WAL 解码输出（wal2json）
SELECT * FROM pg_logical_slot_get_changes('slot_demo', NULL, NULL, 'format_version', '2');
-- 输出示例：{"change":[{"kind":"insert","schema":"public","table":"orders","columnnames":["id","amount"],...}]}

Publisher (主库)
  └── WAL 写入 → 逻辑解码 → 复制槽 (slot)
                         │  wal_sender 进程
                         ▼
                网络 (TCP)
                         │
                         ▼
Subscriber (订阅库)
  └── 接收变更 → 应用到表

-- 在源库（Publisher）上创建发布
CREATE PUBLICATION mypub FOR TABLE orders, users;

-- 发布全部表
CREATE PUBLICATION all_tables FOR ALL TABLES;

-- 只发布特定操作
CREATE PUBLICATION orders_insert_only
  FOR TABLE orders WITH (publish = 'insert');

-- 在目标库（Subscriber）上创建订阅
CREATE SUBSCRIPTION mysub
  CONNECTION 'host=primary.example.com port=5432 dbname=mydb user=repl password=xxx'
  PUBLICATION mypub
  WITH (create_slot = true, enabled = true);

-- 查看发布与订阅
SELECT * FROM pg_publication;
SELECT * FROM pg_subscription;
SELECT * FROM pg_stat_subscription;

-- 暂停/恢复/删除
ALTER SUBSCRIPTION mysub DISABLE;
ALTER SUBSCRIPTION mysub ENABLE;
DROP SUBSCRIPTION mysub;

-- 修改订阅连接参数
ALTER SUBSCRIPTION mysub CONNECTION 'host=newhost ...';

1. 在主库创建复制槽
2. 对发布表执行基础快照（pg_basebackup 类似，但按表）
3. 将快照数据 COPY 到订阅端
4. 快照之后的新变更通过 WAL 解码持续同步

-- 目标库：创建订阅但不自动复制
CREATE SUBSCRIPTION mysub
  CONNECTION '...' PUBLICATION mypub
  WITH (copy_data = false);

-- 用 pg_dump 只导数据（不导 schema，避免覆盖订阅表）
pg_dump -h primary -d mydb --data-only -t orders -t users > data.sql
psql -d target -f data.sql

-- 重新启用订阅开始接续变更
ALTER SUBSCRIPTION mysub ENABLE;

-- 观察复制状态
SELECT subname, received_lsn, latest_end_lsn, last_msg_send_time
FROM pg_stat_subscription;
-- received_lsn 滞后于 latest_end_lsn 说明还有积压

-- 无主键表：让副本记录完整旧行（开销较大）
ALTER TABLE orders REPLICA IDENTITY FULL;

-- 有唯一索引但无主键：用唯一索引
ALTER TABLE orders REPLICA IDENTITY USING INDEX orders_uidx;

-- 数据一致性校验（简单比对行数）
SELECT count(*) FROM orders ON primary;
SELECT count(*) FROM orders ON subscriber;
-- 更严格：用 checksum 按 id 分片比对

-- 1. 在目标新库建好 schema（schema 不自动复制）
-- 2. 源库发布
CREATE PUBLICATION mig_pub FOR ALL TABLES;

-- 3. 目标库订阅（自动初始同步 + 持续变更）
CREATE SUBSCRIPTION mig_sub
  CONNECTION 'host=old-db ...' PUBLICATION mig_pub;

-- 4. 观察追赶进度，直到 received_lsn 追上主库 latest_end_lsn
SELECT received_lsn, latest_end_lsn,
       pg_wal_lsn_diff(latest_end_lsn, received_lsn) AS lag_bytes
FROM pg_stat_subscription;

PostgreSQL 主库
   │ WAL
   ▼
Debezium PostgreSQL Connector
   │ 解码变更事件
   ▼
Kafka Topic (orders.cdc)
   │
   ├─ 数据湖 / 数仓（实时入湖）
   ├─ 事件驱动微服务
   └─ 实时搜索/分析（Elasticsearch）

# debezium postgres connector 配置片段
{
  "name": "orders-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "primary.example.com",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "secret",
    "database.dbname": "mydb",
    "database.server.name": "pg-primary",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_slot",
    "publication.name": "dbz_publication",
    "table.include.list": "public.orders,public.users",
    "heartbeat.interval.ms": "5000",
    "topic.prefix": "cdc"
  }
}

# 消费 Kafka 中的 CDC 事件（orders 表的 UPDATE）
kafka-console-consumer.sh \
  --bootstrap-server kafka:9092 \
  --topic cdc.public.orders \
  --from-beginning

# 事件 payload 示例（wal2json 风格）
# {"op":"u","ts_ms":1720000000000,"before":{"id":1,"amount":90},"after":{"id":1,"amount":80}}

-- 订阅端：查看复制进度与延迟
SELECT subname, pid, received_lsn,
       latest_end_lsn,
       pg_wal_lsn_diff(latest_end_lsn, received_lsn) AS lag_bytes,
       last_msg_send_time
FROM pg_stat_subscription;

-- 主库：查看复制槽消费位置
SELECT slot_name, database, active,
       restart_lsn,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS wal_retained
FROM pg_replication_slots;

-- 找出 WAL 堆积最多的复制槽
SELECT slot_name, active,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained
FROM pg_replication_slots
ORDER BY pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn) DESC;

-- 确认无用的槽及时删除
SELECT pg_drop_replication_slot('stale_slot');

# Prometheus 告警：复制延迟
groups:
  - name: pg-logical-replication
    rules:
      - alert: ReplicationLagHigh
        expr: pg_stat_subscription_lag_bytes > 104857600   # 100MB
        for: 5m
        labels: { severity: warning }
      - alert: ReplicationSlotWALRetention
        expr: pg_replication_slots_retained_wal_bytes > 10737418240  # 10GB
        for: 10m
        labels: { severity: critical }

-- 订阅端：关闭同步提交加速 apply
ALTER SUBSCRIPTION mysub SET (synchronous_commit = off);

-- 主库：必要时提升 max_wal_senders / max_replication_slots
-- 调整 WAL 级参数需要重启

-- 加密复制链路示例
CREATE SUBSCRIPTION mysub
  CONNECTION 'host=primary.example.com port=5432 dbname=mydb user=repl sslmode=require'
  PUBLICATION mypub;