七月 1, 2025
摘要：在本教程中，您将学习如何通过转储/重新加载某些表，并基于转储时间来启动逻辑复制。
目录
初始设置
我们在这里使用 PostgreSQL 13 版本的源数据库。在这个简单的演示中，我们将使用 pgbench 初始化一个简单的数据架构，并使用这些标准的表进行复制：
$ pgbench -i -s 10
dropping old tables...
creating tables...
generating data...
100000 of 1000000 tuples (10%) done (elapsed 0.10 s, remaining 0.93 s)
200000 of 1000000 tuples (20%) done (elapsed 0.33 s, remaining 1.31 s)
300000 of 1000000 tuples (30%) done (elapsed 0.52 s, remaining 1.22 s)
400000 of 1000000 tuples (40%) done (elapsed 0.74 s, remaining 1.10 s)
500000 of 1000000 tuples (50%) done (elapsed 0.96 s, remaining 0.96 s)
600000 of 1000000 tuples (60%) done (elapsed 1.20 s, remaining 0.80 s)
700000 of 1000000 tuples (70%) done (elapsed 1.44 s, remaining 0.62 s)
800000 of 1000000 tuples (80%) done (elapsed 1.61 s, remaining 0.40 s)
900000 of 1000000 tuples (90%) done (elapsed 1.74 s, remaining 0.19 s)
1000000 of 1000000 tuples (100%) done (elapsed 1.98 s, remaining 0.00 s)
vacuuming...
creating primary keys...
done.
postgres=# \d
List of relations
Schema | Name | Type | Owner
--------+------------------+-------+----------
public | pgbench_accounts | table | postgres
public | pgbench_branches | table | postgres
public | pgbench_history | table | postgres
public | pgbench_tellers | table | postgres
(4 rows)
假设这些表里面，有个大表的复制经常出问题，也就是 pgbench_accounts。其他三个表复制正常，我们可以把它们放到同一个发布中：
CREATE PUBLICATION pub_test FOR TABLE pgbench_branches, pgbench_history, pgbench_tellers;
SELECT * FROM pg_publication;
pubname | pubowner | puballtables | pubinsert | pubupdate | pubdelete | pubtruncate
----------+----------+--------------+-----------+-----------+-----------+-------------
pub_test | 10 | f | t | t | t | t
(1 row)
在目标实例上，本例中为 PostgreSQL 15，我们需要准备与源实例相同的表结构。您可以使用 pg_dump 仅创建其中的表结构：
$ pg_dump --schema-only postgres > db.sql
$ psql -d postgres -f db.sql
设置逻辑复制
一旦我们有了表结构，我们就可以创建订阅，以附加到源实例上刚刚创建的发布，并等待初始的快照/加载完成：
CREATE SUBSCRIPTION sub_test CONNECTION 'host=localhost port=5435 user=postgres dbname=postgres' PUBLICATION pub_test;
SELECT * FROM pg_subscription;
oid | subdbid | subskiplsn | subname | subowner | subenabled | subbinary | substream | subtwophasestate | subdisableonerr | >
-------+---------+------------+----------+----------+------------+-----------+-----------+------------------+-----------------+----->
16589 | 16559 | 0/0 | sub_test | 10 | t | f | f | d | f | host>
(1 row)
SELECT count(*) FROM pgbench_branches;
count
-------
10
(1 row)
三个表的同步已启动并运行，更改会实时同步到目标实例。通过在源表上插入一行即可轻松验证这一点：
INSERT INTO pgbench_branches VALUES (-1, -1, 'aa');
并检查目标库中的同一行：
SELECT * FROM pgbench_branches WHERE bid = -1;
bid | bbalance | filler
-----+----------+------------------------------------------------------------------------------------------
-1 | -1 | aa
(1 row)
使用 pg_dump 启动复制
现在我们需要处理剩下的表。第一步是在源库中为该表创建一个发布：
CREATE PUBLICATION pub_test_2 FOR TABLE pgbench_accounts;
SELECT * FROM pg_publication;
pubname | pubowner | puballtables | pubinsert | pubupdate | pubdelete | pubtruncate
------------+----------+--------------+-----------+-----------+-----------+-------------
pub_test | 10 | f | t | t | t | t
pub_test_2 | 10 | f | t | t | t | t
(2 rows)
为了能够从特定时间点开始复制，我们需要在源数据库上创建一个快照作为起始点。我们可以创建一个到源数据库的复制连接，然后定义一个逻辑复制槽来做到。这里需要注意的是，您需要保持此连接打开，直到复制完全建立。否则，您会丢失快照：
$ psql "dbname=postgres replication=database"
psql (13.12)
Type "help" for help.
postgres=# CREATE_REPLICATION_SLOT my_logical_repl_slot LOGICAL pgoutput;
slot_name | consistent_point | snapshot_name | output_plugin
----------------------+------------------+---------------------+---------------
my_logical_repl_slot | 0/37404478 | 00000003-00000097-1 | pgoutput
(1 row)
这为我们提供了可以使用 pg_dump 进行导出的快照（00000003-00000097-1）：
$ pg_dump --snapshot=00000003-00000097-1 -a -t public.pgbench_accounts > pgbench_accounts.sql
在目标库上我们将加载 pgbench_accounts 表：
postgres=# \i pgbench_accounts.sql
postgres=# select count(*) from public.pgbench_accounts;
count
---------
1000000
(1 row)
此时，至少在实际情况下，源数据库的 pgbench_accounts 表中仍在发生更改，而目标数据库尚未收到这些更改。现在我们需要一个新的订阅，该订阅将附加到我们在复制连接中创建的逻辑复制槽。这里需要注意的是，该订阅不会自动加载初始数据，您可以使用以下选项指定：
CREATE SUBSCRIPTION sub_test_2
CONNECTION 'host=localhost port=5435 user=postgres dbname=postgres'
PUBLICATION pub_test_2 WITH (slot_name='my_logical_repl_slot', create_slot='false' , enabled='false', copy_data='false');
完成这些之后，我们就可以开始复制了：
ALTER SUBSCRIPTION sub_test_2 ENABLE;
验证
现在，在源数据库上添加数据，应该会触发将新数据复制到目标数据库：
INSERT INTO pgbench_accounts
SELECT i,i,i,i::text FROM generate_series(1000001, 1000100) i;
在目标数据库上，我们应该会在 pgbench_accounts 表中看到增加了 100 行：
SELECT count(*) FROM public.pgbench_accounts;
count
---------
1000100
(1 row)
运行正常。如果您的表由于某种原因，无法在项目可接受的时间内完成初始复制，这可能是一个不错的解决方法。