PostgreSQL 逻辑复制与数据同步：从发布订阅到零信任双向同步

一、为什么需要逻辑复制

某客户的 HA-System 用 PostgreSQL 9.2，业务要求：主库在线交易，备库做报表查询——但不允许备库有写操作（防止主备数据不一致）。物理复制（streaming replication）把整个实例复制到备库，备库只能读、不能写任何表，连临时表都不行。

这就是物理复制的限制：备库是主库的完整镜像，粒度是整个实例。如果你想：

• 只复制部分表（不同步日志表）
• 备库也能写入（双向同步）
• 复制到不同版本的 PostgreSQL
• 复制到其他数据库（MySQL、Kafka 等）

物理复制做不到——这些都是逻辑复制的核心场景。

经验法则：物理复制用于 HA（主备切换），逻辑复制用于数据分发（业务拆分、报表分流、CDC 管道）。两者互补，不替代。

二、逻辑复制 vs 物理复制

              物理复制                    逻辑复制
              (Streaming)               (Logical)
                
  主库 ─── WAL 流 ───→ 备库        主库 ─── 逻辑解码 ───→ 订阅端
  (字节级完整复制)                (SQL 语句级复制)
  
  复制粒度: 整个实例               复制粒度: 单表/单数据库
  备库状态: 只读                   订阅端: 可读写
  版本要求: 必须同版本             版本要求: 可跨版本
  DDL 同步: 自动                  DDL 同步: 需手动
  延迟: 通常 < 1s                 延迟: 通常 1~5s
  适用: HA 主备切换               适用: 数据分发/汇聚/同步

三、逻辑复制架构：发布与订阅

  Publisher (主库)                    Subscriber (订阅端)
  ┌──────────────┐                   ┌──────────────┐
  │ Publication   │─── 逻辑解码 ──→ │ Subscription  │
  │ (发布端)      │                   │ (订阅端)      │
  │              │                   │              │
  │ ALTER PUBLI- │                   │ CREATE SUBS- │
  │ CATION pub   │                   │ CRIPTION sub │
  │ FOR TABLE    │                   │ CONNECTION   │
  │ t1, t2, t3   │                   │ = 'host=pub' │
  └──────────────┘                   └──────────────┘
       │                                  │
       │ WAL → 逻辑解码输出                │ 接收 + 应用
       │ (INSERT/UPDATE/DELETE)            │ (本地执行 SQL)
       ▼                                  ▼
  pgoutput 插件                    应用到订阅端表

核心概念：

• Publication（发布）：定义"哪些表的哪些变化要发布出去"
• Subscription（订阅）：定义"从哪个发布端订阅哪些数据"
• pgoutput 插件：PostgreSQL 内置的逻辑解码输出插件，把 WAL 转成逻辑变更（INSERT/UPDATE/DELETE）

四、配置逻辑复制（单向同步）

4.1 发布端配置

-- 发布端（主库）配置

-- 1. 设置 WAL 级别为 logical
-- /etc/postgresql/16/main/postgresql.conf
wal_level = logical
max_replication_slots = 10
max_wal_senders = 10

-- 重启 PostgreSQL 使 wal_level 生效
-- pg_ctl restart -D /data/postgresql/data

-- 2. 创建复制用户
CREATE ROLE replicator LOGIN REPLICATION;
ALTER ROLE replicator SET search_path TO public;

-- 3. 创建发布（publication）
-- 方式一：发布所有表
CREATE PUBLICATION pub_all FOR ALL TABLES;

-- 方式二：发布指定表
CREATE PUBLICATION pub_sales FOR TABLE 
  sales_log, 
  customer_info, 
  product_catalog;

-- 方式三：发布指定表+只发布 INSERT 和 UPDATE（不发布 DELETE）
CREATE PUBLICATION pub_incremental FOR TABLE 
  sales_log, customer_info
  WITH (publish = 'insert, update');

-- 4. 查看发布状态
SELECT * FROM pg_publication;
SELECT * FROM pg_publication_tables;

4.2 订阅端配置

-- 订阅端（报表库）配置

-- 1. 创建与发布端相同的表结构
-- （逻辑复制不同步 DDL，表结构必须预先手动创建）
CREATE TABLE sales_log (
  id SERIAL PRIMARY KEY,
  product_name TEXT,
  quantity INT,
  price NUMERIC(10,2),
  sale_time TIMESTAMP DEFAULT NOW()
);

-- 注意：表结构必须完全匹配，包括列名、类型、顺序
-- 如果发布端有主键，订阅端也必须有（逻辑复制用主键定位行）

-- 2. 创建订阅
CREATE SUBSCRIPTION sub_sales
  CONNECTION 'host=10.10.4.17 port=5432 dbname=production user=replicator password=ReplicatorPass'
  PUBLICATION pub_sales;

-- 3. 初始数据同步
-- 创建订阅时默认会做一次初始同步（复制现有数据）
-- 可关闭：WITH (copy_data = false)

-- 4. 查看订阅状态
SELECT * FROM pg_subscription;
SELECT * FROM pg_subscription_rel;

-- 状态说明：
--   i = 初始化同步中
--   r = 正常复制中（ready）
--   d = 同步完成
--   e = 错误

4.3 监控复制状态

-- 发布端：查看复制槽状态
SELECT 
  slot_name,
  plugin,
  slot_type,
  active,
  restart_lsn,
  confirmed_flush_lsn
FROM pg_replication_slots;

-- 订阅端：查看复制延迟
SELECT 
  subname,
  pid,
  received_lsn,
  latest_end_lsn,
  latest_end_time,
  NOW() - latest_end_time AS replication_lag
FROM pg_stat_subscription;

-- 关键指标：
--   replication_lag < 5s = 正常
--   replication_lag > 30s = 需排查
--   replication_lag > 5min = 严重延迟

4.4 复制槽管理

-- 复制槽防止 WAL 在订阅端未消费前被清理
-- 如果订阅端长期断开，WAL 会堆积导致磁盘满

-- 查看堆积量
SELECT 
  slot_name,
  active,
  pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS wal_retained
FROM pg_replication_slots;

-- 手动清理不活跃的复制槽（订阅端已永久离线）
SELECT pg_drop_replication_slot('sub_sales');

-- 生产建议：设置 wal_keep_size 防止 WAL 被过早清理
-- postgresql.conf
-- wal_keep_size = '1GB'

五、表级选择性同步

5.1 只同步核心表

-- 生产库有 200 张表，但报表只需要 10 张
-- 不要用 FOR ALL TABLES，精确指定需要的表

CREATE PUBLICATION pub_reports FOR TABLE
  sales_log,         -- 销售流水
  customer_info,     -- 客户信息
  product_catalog,   -- 产品目录
  order_header,      -- 订单头
  order_detail,      -- 订单明细
  inventory_stock,   -- 库存
  payment_record,    -- 支付记录
  refund_request,    -- 退款
  daily_summary,     -- 日汇总
  monthly_summary;   -- 月汇总

-- 不同步的表（日志表、临时表、大字段表）：
--   app_log（日均 100 万行，报表不需要）
--   temp_import（临时导入表）
--   audit_trail（审计日志，只在主库查）

5.2 动态增减发布表

-- 追加一张表到发布
ALTER PUBLICATION pub_sales ADD TABLE new_table;

-- 从发布中移除一张表
ALTER PUBLICATION pub_sales DROP TABLE old_table;

-- 订阅端需要刷新才能感知变化
ALTER SUBSCRIPTION sub_sales REFRESH PUBLICATION;

5.3 行级过滤（PostgreSQL 15+）

-- 只发布满足条件的行（行级过滤，PG15 新特性）
CREATE PUBLICATION pub_active_orders FOR TABLE
  order_header
  WHERE (status != 'cancelled'),      -- 只同步非取消的订单
  order_detail
  WHERE (quantity > 0);               -- 只同步有效明细

-- 列级过滤（只同步部分列）
-- 注意：PG15 不支持列级过滤，需要用视图变通
CREATE VIEW v_customer_public AS
  SELECT id, name, region FROM customer_info;  -- 不暴露 email/phone

六、双向复制（Bidirectional Replication）

6.1 配置双向复制

双向复制让两个 PostgreSQL 实例互为发布端和订阅端，各自可读写，变更双向同步。核心风险是冲突——两端同时修改同一行时，谁赢？

-- A 端（10.10.4.17）
CREATE PUBLICATION pub_a FOR TABLE sales_log, customer_info;
CREATE SUBSCRIPTION sub_from_b
  CONNECTION 'host=10.10.4.20 dbname=production user=replicator'
  PUBLICATION pub_b
  WITH (copy_data = false, origin = none);  -- origin=none 防止循环复制

-- B 端（10.10.4.20）
CREATE PUBLICATION pub_b FOR TABLE sales_log, customer_info;
CREATE SUBSCRIPTION sub_from_a
  CONNECTION 'host=10.10.4.17 dbname=production user=replicator'
  PUBLICATION pub_a
  WITH (copy_data = false, origin = none);

origin = none 是关键参数——它告诉订阅端只接收"本地产生"的变更，忽略"从其他节点复制过来的"变更。没有这个参数，变更会在 A→B→A 之间无限循环。

6.2 冲突检测与处理

-- 查看复制冲突
SELECT * FROM pg_stat_subscription_stats;

-- 冲突类型：
--   1. 两侧同时 UPDATE 同一行 → 后到的覆盖先到的（last-writer-wins）
--   2. 一侧 DELETE，另一侧 UPDATE → DELETE 先到则行消失，UPDATE 失败报错
--   3. 两侧 INSERT 同一主键 → 后到的 INSERT 报唯一约束冲突

-- 处理策略：
-- 方案一：业务层面分区（每个节点只写自己分区的数据，避免冲突）
-- 方案二：用触发器记录冲突行到冲突表
-- 方案三：定期比对两端数据，自动修平

6.3 冲突触发器

-- 冲突记录触发器（在订阅端执行）
CREATE TABLE replication_conflicts (
  conflict_time TIMESTAMP DEFAULT NOW(),
  table_name TEXT,
  pk_value TEXT,
  operation TEXT,
  local_data JSONB,
  remote_data JSONB,
  resolved TEXT DEFAULT 'pending'
);

-- 对每张需要冲突检测的表加触发器
CREATE OR REPLACE FUNCTION log_insert_conflict()
RETURNS TRIGGER AS $$
BEGIN
  -- 如果 INSERT 时主键冲突（唯一约束违反）
  -- PostgreSQL 逻辑复制会报错并暂停订阅
  -- 需要在订阅端先解决冲突再恢复订阅
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;

-- 恢复暂停的订阅
ALTER SUBSCRIPTION sub_from_b DISABLE;
-- 手动解决冲突后
ALTER SUBSCRIPTION sub_from_b ENABLE;

七、跨版本与跨数据库同步

7.1 跨版本逻辑复制

-- PostgreSQL 逻辑复制支持跨版本：
-- 发布端 PG 16 → 订阅端 PG 14 ✓（向下兼容）
-- 发布端 PG 14 → 订阅端 PG 16 ✓（向上兼容）

-- 限制：表结构必须完全匹配
-- PG16 新增的列在 PG14 订阅端不存在 → 复制失败
-- 解决：订阅端也加同样列（即使 PG14 不使用该特性）

-- 例如：PG16 发布端用了 GENERATED ALWAYS AS_identity 列
-- PG14 订阅端不支持 → 手动加普通列 + 触发器模拟

7.2 CDC 管道：PostgreSQL → Kafka

逻辑解码可以把变更输出到 Kafka，实现 CDC（Change Data Capture）管道：

# 安装 Debezium PostgreSQL Connector
# Kafka Connect 配置

{
  "name": "pg-source-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "10.10.4.17",
    "database.port": "5432",
    "database.user": "replicator",
    "database.dbname": "production",
    "slot.name": "debezium_slot",
    "plugin.name": "pgoutput",
    "publication.name": "pub_all",
    "database.history.kafka.topic": "schema-changes",
    "database.history.kafka.bootstrap.servers": "kafka:9092",
    
    "table.include.list": "public.sales_log,public.customer_info",
    
    "topic.prefix": "pg_production"
  }
}

# Kafka Topic 输出：
# pg_production.public.sales_log  — 每条变更是一条消息
# 消息格式：
# {
#   "before": {"id": 1, "product_name": "旧值"},
#   "after":  {"id": 1, "product_name": "新值"},
#   "op": "u",  // c=create, u=update, d=delete, r=read(snapshot)
#   "ts_ms": 1721027800000
# }

7.3 pglogical（第三方逻辑复制插件）

-- pglogical 比内置发布订阅更灵活（2ndQuadrant 开发）
-- 支持：行级/列级过滤、双向复制冲突策略、多主架构

-- 安装
CREATE EXTENSION pglogical;

-- 发布端
SELECT pglogical.create_node(
  node_name := 'provider_node',
  dsn := 'host=10.10.4.17 dbname=production'
);

SELECT pglogical.create_replication_set(
  set_name := 'sales_set',
  replicate_insert := true,
  replicate_update := true,
  replicate_delete := true
);

SELECT pglogical.replication_set_add_table(
  set_name := 'sales_set',
  relation := 'sales_log',
  row_filter := 'status != ''cancelled'''  -- 行级过滤
);

-- 订阅端
SELECT pglogical.create_node(
  node_name := 'subscriber_node',
  dsn := 'host=10.10.4.20 dbname=reports'
);

SELECT pglogical.create_subscription(
  subscription_name := 'sales_sub',
  provider_dsn := 'host=10.10.4.17 dbname=production',
  replication_sets := '{sales_set}'
);

八、零信任架构下的多源数据汇聚

8.1 架构设计

  ┌──── 数据源 A ────┐    ┌──── 数据源 B ────┐    ┌──── 数据源 C ────┐
  │ PostgreSQL (HA)  │    │ MySQL (业务)     │    │ Oracle (ERP)     │
  │ 10.10.4.17       │    │ 10.10.4.18       │    │ 10.10.4.19       │
  └──────────┬───────┘    └──────────┬───────┘    └──────────┬───────┘
             │                       │                       │
       逻辑复制                Debezium CDC            Debezium CDC
             │                       │                       │
             ▼                       ▼                       ▼
  ┌──────────────────────────────────────────────────────────────┐
  │                  汇聚数据库 (PostgreSQL 16)                    │
  │                       10.10.4.20                              │
  │  ┌─────────┐ ┌─────────┐ ┌─────────┐                        │
  │  │ ha_tab  │ │ biz_tab │ │ erp_tab │  ← 各源数据分区        │
  │  └─────────┘ └─────────┘ ┌─────────┘                        │
  │              ┌──────────────────────┐                        │
  │              │  统一查询层 (VIEW)    │  ← 跨源关联查询       │
  │              └──────────────────────┘                        │
  └──────────────────────────────────────────────────────────────┘
             │
        mTLS + 证书认证

8.2 安全加固：mTLS 连接

-- 逻辑复制连接使用 SSL 证书认证（而非密码）
CREATE SUBSCRIPTION sub_secure
  CONNECTION 'host=10.10.4.17 port=5432 dbname=production 
              sslmode=verify-full 
              sslcert=/etc/ssl/certs/sub.crt 
              sslkey=/etc/ssl/certs/sub.key 
              sslrootcert=/etc/ssl/certs/ca.crt 
              user=replicator'
  PUBLICATION pub_sales;

-- 发布端 pg_hba.conf 只允许证书认证的连接
# hostssl replication replicator 10.10.4.20/32 cert map=replicator_map

-- pg_ident.conf
# replicator_map  /CN=sub.scrcx.cn  replicator

8.3 数据分区与视图

-- 汇聚库中，不同数据源的表用 schema 分区
CREATE SCHEMA ha_source;    -- HA-System 数据
CREATE SCHEMA biz_source;   -- 业务 MySQL 数据
CREATE SCHEMA erp_source;   -- ERP Oracle 数据

-- 统一查询视图
CREATE VIEW v_unified_sales AS
SELECT 
  'HA' AS source_system,
  sale_time,
  product_name,
  quantity,
  price
FROM ha_source.sales_log
UNION ALL
SELECT 
  'BIZ' AS source_system,
  sale_time,
  product_name,
  quantity,
  price
FROM biz_source.sales_log;

-- 权限控制：不同角色只能访问对应 schema
GRANT USAGE ON SCHEMA ha_source TO ha_app_role;
GRANT SELECT ON ALL TABLES IN SCHEMA ha_source TO ha_app_role;

九、性能调优

9.1 批量发送优化

-- postgresql.conf（发布端和订阅端都调）
-- 逻辑复制默认单行发送，大事务时效率低

wal_level = logical
max_replication_slots = 10
max_wal_senders = 10

-- 订阅端批量应用（PG14+）
-- logical_decoding_work_mem = '64MB'  -- 逻辑解码内存缓冲
-- 更大的缓冲 → 更多变更打包发送 → 减少网络往返

-- 发布端 WAL 发送优化
wal_sender_delay = 10ms          -- 低延迟场景设小值
max_wal_size = '2GB'             -- 保留更多 WAL 防堆积

9.2 订阅端并行应用

-- PG16+ 支持订阅端并行应用逻辑复制消息

-- 创建订阅时指定并行度
CREATE SUBSCRIPTION sub_parallel
  CONNECTION 'host=10.10.4.17 ...'
  PUBLICATION pub_sales
  WITH (streaming = on,   -- 启用流式并行
        synchronous_commit = off);  -- 异步提交

-- streaming = on 时，大事务在订阅端并行 apply
-- 小事务仍然串行保证顺序

-- 查看并行应用状态
SELECT * FROM pg_stat_subscription_stats;

9.3 网络与连接池

-- 逻辑复制是长连接，网络稳定性至关重要
-- 发布端 pg_hba.conf 设置连接超时

# hostssl replication replicator 10.10.4.20/32 md5
# 连接失败自动重试，默认 5s 间隔，可调整

-- 订阅端设置重连参数
-- PostgreSQL 16+ 支持：
-- subconninfo 里加 keepalive 参数
CREATE SUBSCRIPTION sub_reliable
  CONNECTION 'host=10.10.4.17 port=5432 ... 
              keepalives=1 
              keepalives_idle=30 
              keepalives_interval=10 
              keepalives_count=5'
  PUBLICATION pub_sales;

十、故障排查

10.1 常见问题诊断

-- 1. 订阅端报 "could not receive data from WAL stream"
-- 原因：发布端 WAL 已清理，订阅端落后太多
-- 解决：重新初始化订阅
ALTER SUBSCRIPTION sub_sales DISABLE;
ALTER SUBSCRIPTION sub_sales REFRESH PUBLICATION;
ALTER SUBSCRIPTION sub_sales ENABLE;
-- 或重新创建订阅（copy_data=true 重新全量同步）

-- 2. 订阅端报 "unique constraint violation"
-- 原因：订阅端已有主键冲突的数据
-- 解决：先删除冲突行，再恢复订阅
DELETE FROM sales_log WHERE id = <冲突ID>;
ALTER SUBSCRIPTION sub_sales ENABLE;

-- 3. 复制延迟突然增大
-- 原因排查：
SELECT 
  now() - pg_last_xact_replay_timestamp() AS replay_lag
FROM pg_stat_subscription;
-- 如果 replay_lag > 5min：
--   检查网络带宽
--   检查订阅端是否大事务阻塞
--   检查发布端 WAL 是否堆积（pg_replication_slots）

10.2 WAL 堆积处理

-- 如果复制槽长时间不活跃，WAL 堆积导致磁盘空间告警：

-- 1. 查看堆积量
SELECT slot_name, active,
  pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_insert_lsn(), restart_lsn)) AS retained
FROM pg_replication_slots;

-- 2. 如果订阅端已永久不需要，删除复制槽
SELECT pg_drop_replication_slot('sub_sales');

-- 3. 紧急释放空间（生产慎用）
-- 设置 wal_keep_size = 0，让 PostgreSQL 自动清理不需要的 WAL
-- 但要确保其他复制槽不会被影响

-- 4. 最佳做法：设置监控告警
-- 当 WAL 堆积 > 1GB 时告警
-- 当复制槽不活跃 > 24 小时时告警

十一、避坑清单

#	坑	现象	解决
1	wal_level 未设 logical	创建发布报错	ALTER SYSTEM SET wal_level = logical; 重启生效
2	表结构不匹配	订阅端报列类型不一致	两端用相同 CREATE TABLE 或 pg_dump 同步 DDL
3	无主键的表	逻辑复制无法定位行	所有复制表必须有主键或 REPLICA IDENTITY
4	订阅端表已有数据	初始同步报唯一约束冲突	WITH (copy_data = false) 或先清空订阅端表
5	DDL 不自动同步	发布端加列后订阅端报错	手动在订阅端执行 DDL，或用事件触发器通知
6	双向复制循环	变更 A→B→A 无限传播	WITH (origin = none) 阻断循环
7	复制槽堆积 WAL	发布端磁盘空间告警	监控 pg_replication_slots，设置 wal_keep_size
8	网络断开导致延迟	订阅端落后数小时	keepalive 参数 + 订阅自动重连
9	大事务阻塞复制	一条 UPDATE 百万行	订阅端开启 streaming = on 并行应用
10	REPLICA IDENTITY 不够	UPDATE/DELETE 复制失败	ALTER TABLE t REPLICA IDENTITY FULL;（无主键时）

十二、小结

逻辑复制解决了物理复制做不到的三件事：选择性同步、双向写入、跨版本/跨数据库。但它也带来了额外复杂度——DDL 不自动同步、冲突需要业务层处理、WAL 堆积需要监控。

核心原则三条：

1. 精确发布：只发布需要的表和操作，不要 FOR ALL TABLES——减少网络带宽、降低冲突概率。
2. 监控 WAL 堆积：复制槽是逻辑复制的生命线，不活跃的槽会导致磁盘空间耗尽——监控是必须的，不是可选的。
3. DDL 手动同步：逻辑复制不传 DDL，任何结构变更必须两端同步执行。可以用事件触发器自动化通知，但执行仍需手动。

当你第一次看到只同步 10 张核心表到报表库、报表查询不再拖慢主库、变更在 3 秒内到达时，你就理解了"逻辑复制不是物理复制的替代品，而是数据分发的利器"这句话。