PostgreSQL 逻辑复制的第一步

大多数应用程序最初都从单个 PostgreSQL 数据库开始，但随着时间的推移，扩展、分发负载或集成的需求自然会产生。PostgreSQL 的逻辑复制是满足这些需求的功能之一，它通过发布-订阅模型将一个 PostgreSQL 实例中的行级更改流式传输到另一个实例。逻辑复制不仅仅是一个高级功能，它提供了一个灵活的框架，你可以基于它进一步在你的架构中分发和集成 PostgreSQL。

在本文中，我们将从基础开始，探索逻辑复制背后的核心理念，并学习如何使用它。

物理复制与逻辑复制

在深入探讨之前，让我们了解 PostgreSQL 中复制的作用以及它是如何基于预写日志（Write-Ahead Log，WAL）构建的。

WAL 是一个顺序的、追加式的日志，记录了对集群数据所做的每一次更改。为了确保数据的持久性，所有修改首先写入 WAL，然后才永久写入磁盘。这使得 PostgreSQL 可以通过重放日志中的更改来从崩溃中恢复。

并发事务所需的版本化更改是通过多版本并发控制（Multi-Version Concurrency Control，MVCC）管理的。MVCC 不直接覆盖数据，而是创建行的多个版本，允许每个事务看到数据库的一致快照。正是 WAL 捕获了这些版本化更改以及事务元数据，以确保在任何给定时间点的数据一致性。

物理复制直接基于预写日志构建。它允许将主服务器上的二进制 WAL 数据流式传输到一个或多个备用服务器（副本），从而有效地创建整个集群的逐字节副本。这一要求使得副本是只读的，使它们成为故障转移或扩展用途的理想选择。

与之相比，逻辑复制虽然也是基于 WAL 数据构建的，但采用了根本不同的方法。逻辑复制不是直接流式传输原始更改数据，而是将 WAL 解码为逻辑的行级更改（如 INSERT、UPDATE 和 DELETE），然后通过发布-订阅模型将它们发送给订阅者。与物理复制相比，逻辑复制允许选择性复制，同时允许可写的订阅者，这些订阅者不严格绑定到单个发布者。这种灵活性虽然增加了可用设置的多样性，但逻辑复制不会复制 DDL 更改。

物理复制是用于高可用性和灾难恢复的首选方案，当你需要一个快速、精确的整个数据库集群副本，以便在发生故障时接管时。它设置简单且非常高效，但灵活性有限。

逻辑复制在你需要对复制的数据进行细粒度控制、需要可写的副本，或者希望将 PostgreSQL 与其他系统或版本集成时表现出色。它适用于零停机升级、多区域部署以及构建可扩展、模块化的架构。

设置逻辑复制

现在我们已经了解了足够的理论知识。要开始设置逻辑复制，你需要两个 PostgreSQL 实例——你可以选择在两台虚拟机上运行，或者在同一台计算机上运行两个集群。我们将其中一个称为发布者，另一个称为订阅者。

准备发布者

首先，你需要准备发布者以发出逻辑更改。你需要修改 postgresql.conf
（或添加特定的 conf.d 配置），并添加以下参数：

wal_level = logical
max_replication_slots = 10
max_wal_senders = 10

发布者还需要能够被订阅者通过 TCP 套接字访问：

listen_addresses = '*'

这里的配置说明如下：

• wal_level 设置为 logical 是配置的关键部分。它告诉 PostgreSQL 要写入 WAL 的信息量，在这种情况下，是为了支持逻辑解码。
• max_replication_slots 定义了可以在服务器上创建的最大复制槽数量。每个逻辑复制订阅都需要自己的槽。
• max_wal_senders 应该设置得足够高，以容纳所有预期的来自订阅者的并发复制连接。每个活跃的复制订阅会占用一个 wal_sender 槽。
  你还需要在 pg_hba.conf 中配置客户端认证，以允许订阅者连接进行复制（为简化起见，我们允许所有用户）：

# TYPE      DATABASE        USER            ADDRESS                 METHOD
host        replication     all             subscriber_ip/32        scram-sha-256

虽然在本文中我们假设使用超级用户，这会比较方便：

CREATE USER my_user_name SUPERUSER PASSWORD 'my_secure_password';

但在实际部署中，建议使用专用的复制用户：

CREATE USER replication_user WITH REPLICATION ENCRYPTED PASSWORD 'my_secure_password';

配置完成后，重启 PostgreSQL 以使配置更改生效。之后，连接到服务器并准备环境。

-- psql 示例
CREATE DATABASE logical_demo_publisher;
\c logical_demo_publisher
创建一个示例表并插入初始数据：

CREATE TABLE products (
   id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
   name TEXT NOT NULL,
   category TEXT,
   price DECIMAL(10, 2) NOT NULL,
   stock_quantity INT DEFAULT 0,
   created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
   updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
   description TEXT,
   is_active BOOLEAN DEFAULT TRUE
);

-- 插入 10 条数据
INSERT INTO products (
   name,
   category,
   price,
   stock_quantity,
   description,
   is_active
)
SELECT
   'Product Batch ' || s.id AS name,
   CASE (s.id % 5)
       WHEN 0 THEN 'Electronics'
       WHEN 1 THEN 'Books'
       WHEN 2 THEN 'Home Goods'
       WHEN 3 THEN 'Apparel'
       ELSE 'Miscellaneous'
   END AS category,
   ROUND((RANDOM() * 500 + 10)::numeric, 2) AS price,
   FLOOR(RANDOM() * 200)::int AS stock_quantity,
   'Auto-generated description for product ID ' || s.id || '. Lorem ipsum dolor sit amet, consectetur adipiscing elit.' AS description,
   (s.id % 10 <> 0) AS is_active
FROM generate_series(1, 10) AS s(id);

发布者的最后一步是创建一个发布，定义我们想要发布哪些数据。

CREATE PUBLICATION my_publication FOR TABLE products;

准备订阅者

接下来，我们将使用订阅者实例来接收逻辑更改。订阅者无需进行任何专门的配置更改，因为它不会发出更改（请注意“专门”）。

创建目标数据库和架构。架构的创建是一个重要的部分：

-- psql 示例
CREATE DATABASE my_subscriber_db;
\c my_subscriber_db

CREATE TABLE products (
   id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
   name TEXT NOT NULL,
   category TEXT,
   price DECIMAL(10, 2) NOT NULL,
   stock_quantity INT DEFAULT 0,
   created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
   updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
   description TEXT,
   is_active BOOLEAN DEFAULT TRUE
);

现在我们已经为测试逻辑复制奠定了基础。最简单的测试方法是使用连接详细信息和要订阅的发布名称创建一个订阅。

CREATE SUBSCRIPTION my_subscription
   CONNECTION 'host=publisher_ip_address port=5432 user=your_replication_user password=my_secure_password dbname=logical_demo_publisher'
   PUBLICATION my_publication;

• my_subscription 应该是对你订阅的描述性名称。
• CONNECTION 定义了如何连接到发布者（这是一个常规的连接字符串，也可以是一个服务）。
• PUBLICATION 指定了你之前在发布者上创建的发布名称。
  如果连接正确，订阅将默认开始初始数据同步并监听传入的更改。如果你使用了上面的查询语句，你可以验证数据现在是否已在订阅服务器上。

# SELECT count(1) FROM products;
 count
---------
    10
(1 row)

记录数应与上面创建的记录数一致generate_series（本例中为 10）。您可以继续在发布服务器实例上再次插入一行，并验证复制到订阅服务器的数据。

恭喜！您已在 PostgreSQL 中设置了第一个逻辑复制！

逻辑复制的核心概念

这很简单，对吧？这就是目标。既然您已经了解了逻辑复制的实际操作，让我们更深入地探讨实现这一目标的核心概念。

发布

顾名思义，发布是一切的起点。它本质上是发布者提供的数据的目录。您可以发布许多对象：

--- specific tables
CREATE PUBLICATION my_publication FOR TABLE products, orders;

--- everything in your database (make sure you really want to do this)
CREATE PUBLICATION all_data FOR ALL TABLES;

--- replicate specific columns only (PostgreSQL 15 and higher)
CREATE PUBLICATION generic_data FOR TABLE products (id, name, price);

--- filter published data (PostgreSQL 15 and higher)
CREATE PUBLICATION active_products FOR TABLE products WHERE (is_active = true);

--- filter different operations
CREATE PUBLICATION my_publication FOR TABLE products, orders
WITH (publish = 'insert');

--- or you can mix & match it to get exactly what you need
CREATE PUBLICATION eu_customers FOR
    TABLE customers (id, email, country, created_at)
        WHERE (country IN ('DE', 'FR', 'IT')),
    TABLE orders (id, customer_id, total_amount)
        WHERE (total_amount > 100)
WITH (publish = 'insert, update');

由此可见，逻辑复制确实提供了相当多的选项，远非物理复制那种僵硬的、逐字节的复制。而这正是允许你构建复杂拓扑的特性。

设置出版物后，您可以进行查看：

--- in psql
\dRp
\dRp+ my_publication

--- using pg_catalog
SELECT * FROM pg_publication_tables WHERE pubname = 'my_publication';

从这些选项中，我们很容易将出版物视为可定制的数据馈送。

订阅

逻辑复制的另一端是订阅。它定义了从发布者消费哪些事件以及如何消费。您可以使用连接详细信息和一些选项来订阅发布。

--- basic subscription with full connection detail
CREATE SUBSCRIPTION my_subscription
    CONNECTION 'host=publisher_host port=5432 user=repl_user password=secret dbname=source_db'
    PUBLICATION my_publication;

-- subscription using service definition
CREATE SUBSCRIPTION realtime_only
    CONNECTION 'service=my_source_db'
    PUBLICATION my_publication;

订阅的默认行为是复制现有数据，立即启动，并继续流式传输数据变更。一旦您开始尝试逻辑复制，就可以控制该行为。

--- subscribe without initial copy of the data
CREATE SUBSCRIPTION streaming_only
    CONNECTION 'service=my_source_db'
    PUBLICATION my_publication
    WITH (copy_data = false);

--- or defer the start for later
CREATE SUBSCRIPTION manual_start
    CONNECTION 'service=my_source_db'
    PUBLICATION my_publication
    WITH (enabled = false);

您可以监控您的订阅：

--- in psql
\dRs
\dRs+ my_subscription

--- or their status using pg_catalog
SELECT subname, received_lsn, latest_end_lsn, latest_end_time
FROM pg_stat_subscription;

复制槽

发布和订阅建立了数据流，但缺少一个关键部分：发布者如何跟踪以不同速度读取 WAL 的多个订阅者？复制槽就是答案。它们在 WAL 流中充当持久预订，精确跟踪每个订阅者当前的位置（或上次的位置），确保即使连接中断也不会丢失任何更改。

让我们看看复制槽可以告诉我们有关其自身的哪些信息。

# SELECT * FROM pg_replication_slots;
-[ RECORD 1 ]-------+---------------
slot_name           | my_subscription
plugin              | pgoutput
slot_type           | logical
datoid              | 16390
database            | my_db_source
temporary           | f
active              | t
active_pid          | 3162
xmin                |
catalog_xmin        | 777
restart_lsn         | 0/1DEFBF0
confirmed_flush_lsn | 0/1DEFC28
wal_status          | reserved
safe_wal_size       |
two_phase           | f
inactive_since      |
conflicting         | f
invalidation_reason |
failover            | f
synced              | f

最有趣的属性是：

• slot_name
• plugin用于逻辑复制
• slot_type确认它用于逻辑复制
• active指示是否有订阅者从此槽读取

其中最重要的是：

• restart_lsn识别此槽位“保留”WAL 文件以防止其被释放的 LSN
• confirmed_flush_lsn这是订户确认已成功处理（即应用）的最后一个 LSN 位置。
  虽然我们不会详细介绍 WAL，但可以说 LSN（日志序列号）是WAL 中唯一的地址或位置，用于标识数据库更改流中的位置。

大多数情况下，您无需手动创建复制槽——订阅会创建和管理它们。复制槽的持久性确保它们在发布者和订阅者重启后依然有效。

请注意，如果订阅者未主动消费更改，则发布者 PostgreSQL 将不会发布包含槽位尚未消费的更改的 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;

-[ RECORD 1 ]+--------------------
slot_name    | my_subscription
active       | t
wal_retained | 265 MB

复制身份

正如我们已经讨论过的，逻辑复制适用于行级更改，例如INSERT、UPDATE和DELETE。并非所有操作都相同。虽然INSERT相对简单（向订阅服务器发送新行），但对于UPDATE和DELETE操作，PostgreSQL 都需要能够唯一地标识要在订阅服务器上修改的目标行。

这由每个已发布表的属性管理REPLICA IDENTITY。默认情况下，如果表有主键（它有，对吧？），或者您可以USING INDEX为唯一索引指定它，或者作为替代方案，指定（通常应避免）所有旧列的行值以查找要更新的目标行。请注意，如果没有主键或合适的键索引，FULLPostgreSQL 可能会默认这样做。REPLICA IDENTITY FULL

只要有可能，请始终：

• 选择REPLICA IDENTITY DEFAULT（用于主键），为您提供最高效的选择。如果您没有主键，请考虑使用它（代理索引始终是一个选择）。
• REPLICA IDENTITY USING INDEX index_name是否存在更好（或更小）的唯一索引来匹配您的数据库模型的业务逻辑或架构。

在复制身份时，有一些特殊情况是不该做的，但我们将在本指南的高级部分中讨论这些情况。

架构变更

尽管已经提到过，但必须重申的是，逻辑复制不会自动复制数据定义语言 (DDL) 更改，因为它仅传输行级更改而不是完整的 WAL 段。

在复制反映更改的数据之前，必须手动将发布者上所做的任何架构更改应用到所有订阅者。

我们将在本指南的后面部分专门介绍模式，但总而言之，这是涉及模式更改时推荐的操作顺序：

1. （可选）但建议对于重大更改，在适用的情况下暂停复制。
2. 应用并验证对订阅者的 DDL 更改。
3. 在发布者上应用并验证 DDL 更改。
4. 如果适用，则恢复复制。

虽然与物理复制相比，这似乎是一个主要缺点，但它却为我们提供了最大的灵活性。

其他考虑因素

由于逻辑复制能够可靠地处理行级更改，因此了解其他具体的限制和行为至关重要。具体来说：

• 重要的是要理解序列是不可复制的。当它们用于在发布者上生成值，并且它们的值在那里递增时，订阅者上相应的序列（如果存在）将不会被更新。
• 其他数据库对象不会被复制。虽然对于视图、存储过程、触发器和规则来说，复制操作或许可以理解，但对于物化视图，你也不能依赖它。
• 必须特别注意用户定义的数据类型（类似于模式更改）。如果复制表使用用户定义类型（例如，使用 的列ENUM），则该类型必须已在订阅服务器上可用，并且具有完全相同的名称和结构。由于ENUMs 表示为有序集合，因此即使值的顺序也很重要！

逻辑复制

在之前的示例中，我们设置了发布者和订阅者，依赖数据的初始副本，并让一切正常运行。它们在重启后都能继续运行，并且由于复制槽的存在，它们一旦上线就能继续运行。

但是，如果您需要执行维护或定期操作（例如在订阅者或发布者上更新架构），该怎么办？有时您可能需要暂时停止复制。PostgreSQL 通过订阅管理简化了这一过程，允许您停止使用更改。

ALTER SUBSCRIPTION my_subscription DISABLE;
正如我们之前提到的，禁用后，订阅将停止使用来自发布者的更改，同时保留复制槽。这意味着：

订阅者未应用任何新的更改。
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;

准备就绪后，您可以恢复订阅。

ALTER SUBSCRIPTION my_subscription ENABLE;

协调架构变更

如前所述，由于逻辑复制不会自动复制 DDL 变更，因此您需要手动协调架构更新。以下是逐步介绍如何执行此类变更的基本概述。

首先，禁用订阅者上的复制并应用架构更改。

--- disable replication
ALTER SUBSCRIPTION my_subscription DISABLE;

--- apply changes on the subscriber
ALTER TABLE products ADD COLUMN category_id INTEGER;
CREATE INDEX products_by_category ON products(category_id);

只有这样，您才能将更改应用到发布者。

ALTER TABLE products ADD COLUMN category_id INTEGER;
CREATE INDEX products_by_category ON products(category_id);

并恢复复制。

ALTER SUBSCRIPTION my_subscription ENABLE;

如果您首先将架构更改应用于发布者，则使用新列的任何新数据都将无法复制到尚没有该列的订阅者。

处理不兼容的变更

PostgreSQL 的默认冲突解决机制非常简单。当冲突发生时，逻辑复制将停止，订阅服务器将进入错误状态。例如，假设订阅服务器上已经存在某条产品记录，您将在日志文件中看到类似如下内容：

ERROR: duplicate key value violates unique constraint "products_pkey"
DETAIL: Key (id)=(452) already exists.
CONTEXT:  processing remote data for replication origin "pg_16444" during message type "INSERT" for replication target relation "public.products" in transaction 783, finished at 0/1E020E8
LOG:  background worker "logical replication apply worker" (PID 457) exited with exit code 1

如您所见，逻辑复制已停止并显示错误代码，除非您手动解决冲突，否则无法继续。为此，您需要识别冲突的数据并确定如何处理它。

-- option 1: remove the conflicting data
DELETE FROM products WHERE id = 452;

-- option 2: update the data to match the expected state
UPDATE products SET name = 'New Product', price = 29.99 WHERE id = 452;

只有解决冲突后，逻辑复制才能继续。这只是逻辑复制过程中可能出现的冲突的一个非常简单的示例。其他示例可能涉及：

• 数据问题、约束违规或类型不匹配
• 架构冲突（缺失或重命名表/列）
• 权限问题或行级安全性

在所有这些情况下，您仍然需要先解决问题，然后才能恢复逻辑复制。

结论

PostgreSQL 中的逻辑复制提供了超越传统物理复制的强大可能性。我们介绍了一些基础知识，可以帮助您开始搭建自己的逻辑复制环境，包括理解物理复制和逻辑复制之间的区别、配置发布者和订阅者，以及如何管理发布、订阅和复制槽等核心组件。

本文是即将发布的《精通 PostgreSQL 中的逻辑复制》指南的一部分。如果您对该主题感兴趣，请考虑订阅以获取最新发布的文章。

原文地址：https://boringsql.com/posts/logication-replication-introduction/
原文作者：Radim Marek