— 创建逻辑复制槽
SELECT pg_create_logical_replication_slot(‘fgedu_slot’, ‘wal2json’);

— 输出结果
slot_name | lsn
—————–+———–
fgedu_slot | 0/1A2B3C8
(1 row)

— 查看复制槽信息
SELECT
slot_name,
slot_type,
plugin,
active,
restart_lsn,
confirmed_flush_lsn
FROM pg_replication_slots
WHERE slot_name = ‘fgedu_slot’;

— 输出结果
slot_name | slot_type | plugin | active | restart_lsn | confirmed_flush_lsn
————-+———–+———+——–+————-+——————–
fgedu_slot | logical | wal2json| f | 0/1A2B3C8 | 0/1A2B3C8
(1 row)

— 创建临时复制槽（会话结束后自动删除）
SELECT pg_create_logical_replication_slot(‘fgedu_temp_slot’, ‘pgoutput’, true);

— 输出结果
slot_name | lsn
—————–+———–
fgedu_temp_slot | 0/1A2B4D0
(1 row)

— 创建逻辑复制用户
CREATE USER fgedu_logical REPLICATION LOGIN PASSWORD ‘fgedu123’;

— 输出结果
CREATE ROLE

— 授权
GRANT CONNECT ON DATABASE fgedudb TO fgedu_logical;
GRANT USAGE ON SCHEMA public TO fgedu_logical;

— 配置pg_hba.conf允许复制连接
— 添加以下行到pg_hba.conf
host replication fgedu_logical 192.168.1.0/24 md5

— 重新加载配置
SELECT pg_reload_conf();

— 输出结果
pg_reload_conf
—————-
t
(1 row)

— 查看所有复制槽
SELECT
slot_name,
slot_type,
plugin,
database,
active,
active_pid,
xmin,
catalog_xmin,
restart_lsn,
confirmed_flush_lsn,
pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn) as lag_bytes
FROM pg_replication_slots;

— 输出结果
slot_name | slot_type | plugin | database | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn | lag_bytes
————–+———–+———+———-+——–+————+——+————–+————-+———————+———–
fgedu_slot | logical | wal2json| fgedudb | f | NULL | | 12345 | 0/1A2B3C8 | 0/1A2B3C8 | 0
fgedu_slot2 | logical | pgoutput| fgedudb | t | 5678 | | 12346 | 0/1A2B4D0 | 0/1A2B4D0 | 1024
(2 rows)

— 监控WAL堆积情况
SELECT
slot_name,
active,
restart_lsn,
confirmed_flush_lsn,
pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) as wal_backlog,
pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) as confirmed_backlog
FROM pg_replication_slots;

— 输出结果
slot_name | active | restart_lsn | confirmed_flush_lsn | wal_backlog | confirmed_backlog
————-+——–+————-+———————+————-+——————-
fgedu_slot | f | 0/1A2B3C8 | 0/1A2B3C8 | 128 MB | 128 MB
fgedu_slot2 | t | 0/1A2B4D0 | 0/1A2B4D0 | 0 bytes | 0 bytes
(2 rows)

— 删除复制槽
SELECT pg_drop_replication_slot(‘fgedu_slot’);

— 输出结果
pg_drop_replication_slot
————————–

(1 row)

— 强制删除活跃的复制槽（需要先终止连接）
SELECT pg_terminate_backend(active_pid), pg_drop_replication_slot(‘fgedu_slot2’)
FROM pg_replication_slots
WHERE slot_name = ‘fgedu_slot2’ AND active;

— 输出结果
pg_terminate_backend | pg_drop_replication_slot
———————-+————————–
t |
(1 row)

— 创建测试表
CREATE TABLE fgedu_test_decode (
id SERIAL PRIMARY KEY,
name VARCHAR(100),
value NUMERIC,
create_time TIMESTAMP DEFAULT NOW()
);

— 插入测试数据
INSERT INTO fgedu_test_decode(name, value) VALUES(‘test1’, 100);
INSERT INTO fgedu_test_decode(name, value) VALUES(‘test2’, 200);

— 输出结果
INSERT 0 1
INSERT 0 1

— 创建复制槽
SELECT pg_create_logical_replication_slot(‘fgedu_decode_slot’, ‘wal2json’);

— 输出结果
slot_name | lsn
——————–+———–
fgedu_decode_slot | 0/1A2B5E0
(1 row)

— 更新数据
UPDATE fgedu_test_decode SET value = 150 WHERE name = ‘test1’;

— 输出结果
UPDATE 1

— 删除数据
DELETE FROM fgedu_test_decode WHERE name = ‘test2’;

— 输出结果
DELETE 1

— 获取变更数据
SELECT lsn, xid, data
FROM pg_logical_slot_get_changes(‘fgedu_decode_slot’, NULL, NULL);

— 输出结果
lsn | xid | data
————+——-+————————————————————————————————————
0/1A2B5E8 | 12347 | {“change”:[{“kind”:”insert”,”schema”:”public”,”table”:”fgedu_test_decode”,”columnnames”:[“id”,”name”,”value”,”create_time”],”columnvalues”:[1,”test1″,100,”2026-04-07 16:30:00.123456″],”columntypes”:[“integer”,”character varying(100)”,”numeric”,”timestamp without time zone”]}]}
0/1A2B6F0 | 12348 | {“change”:[{“kind”:”insert”,”schema”:”public”,”table”:”fgedu_test_decode”,”columnnames”:[“id”,”name”,”value”,”create_time”],”columnvalues”:[2,”test2″,200,”2026-04-07 16:30:00.234567″],”columntypes”:[“integer”,”character varying(100)”,”numeric”,”timestamp without time zone”]}]}
0/1A2B7F8 | 12349 | {“change”:[{“kind”:”update”,”schema”:”public”,”table”:”fgedu_test_decode”,”columnnames”:[“id”,”name”,”value”,”create_time”],”columnvalues”:[1,”test1″,150,”2026-04-07 16:30:00.123456″],”oldkeys”:{“keynames”:[“id”],”keyvalues”:[1],”keytypes”:[“integer”]}}]}
0/1A2B900 | 12350 | {“change”:[{“kind”:”delete”,”schema”:”public”,”table”:”fgedu_test_decode”,”oldkeys”:{“keynames”:[“id”],”keyvalues”:[2],”keytypes”:[“integer”]}}]}
(4 rows)

— 使用pg_logical_slot_peek_changes预览变更（不移动进度）
SELECT lsn, xid, data
FROM pg_logical_slot_peek_changes(‘fgedu_decode_slot’, NULL, NULL);

— 输出结果
(0 rows)

— 创建使用pgoutput的复制槽
SELECT pg_create_logical_replication_slot(‘fgedu_pgoutput_slot’, ‘pgoutput’);

— 输出结果
slot_name | lsn
———————-+———–
fgedu_pgoutput_slot | 0/1A2BA00
(1 row)

— 插入数据
INSERT INTO fgedu_test_decode(name, value) VALUES(‘test3’, 300);

— 输出结果
INSERT 0 1

— 使用pgoutput获取变更（需要指定参数）
SELECT lsn, xid, data
FROM pg_logical_slot_get_changes(
‘fgedu_pgoutput_slot’,
NULL,
NULL,
‘proto_version’, ‘1’,
‘publication_names’, ‘fgedu_pub’
);

— 输出结果
ERROR: publication “fgedu_pub” does not exist

— 创建发布
CREATE PUBLICATION fgedu_pub FOR TABLE fgedu_test_decode;

— 输出结果
CREATE PUBLICATION

— 再次获取变更
SELECT lsn, xid, data
FROM pg_logical_slot_get_changes(
‘fgedu_pgoutput_slot’,
NULL,
NULL,
‘proto_version’, ‘1’,
‘publication_names’, ‘fgedu_pub’
);

— 输出结果
lsn | xid | data
————+——-+———-
0/1A2BB08 | 12351 | B: public.fgedu_test_decode
0/1A2BB10 | 12351 | I: public.fgedu_test_decode[3][test3][300][2026-04-07 16:35:00.123456]
(2 rows)

— 创建CDC配置表
CREATE TABLE fgedu_cdc_config (
id SERIAL PRIMARY KEY,
table_name VARCHAR(100) NOT NULL,
schema_name VARCHAR(100) DEFAULT ‘public’,
operation_types VARCHAR(50) DEFAULT ‘INSERT,UPDATE,DELETE’,
enabled BOOLEAN DEFAULT TRUE,
created_at TIMESTAMP DEFAULT NOW()
);

— 创建CDC变更记录表
CREATE TABLE fgedu_cdc_changes (
id SERIAL PRIMARY KEY,
lsn TEXT,
xid BIGINT,
schema_name VARCHAR(100),
table_name VARCHAR(100),
operation VARCHAR(20),
old_data JSONB,
new_data JSONB,
change_time TIMESTAMP DEFAULT NOW()
);

CREATE INDEX idx_cdc_changes_table ON fgedu_cdc_changes(table_name, change_time);
CREATE INDEX idx_cdc_changes_time ON fgedu_cdc_changes(change_time);

— 插入CDC配置
INSERT INTO fgedu_cdc_config(table_name, schema_name)
VALUES
(‘fgedu_orders’, ‘public’),
(‘fgedu_products’, ‘public’),
(‘fgedu_customers’, ‘public’);

— 创建CDC捕获函数
CREATE OR REPLACE FUNCTION fgedu_cdc_capture()
RETURNS void
AS $$
DECLARE
change_record RECORD;
change_data JSONB;
table_name TEXT;
schema_name TEXT;
operation TEXT;
BEGIN
FOR change_record IN
SELECT lsn, xid, data::jsonb
FROM pg_logical_slot_get_changes(‘fgedu_cdc_slot’, NULL, NULL)
LOOP
change_data := change_record.data;

IF change_data ? ‘change’ THEN
FOR i IN 0..jsonb_array_length(change_data->’change’) – 1 LOOP
table_name := change_data->’change’->i->>’table’;
schema_name := change_data->’change’->i->>’schema’;
operation := UPPER(change_data->’change’->i->>’kind’);

INSERT INTO fgedu_cdc_changes(
lsn, xid, schema_name, table_name, operation,
old_data, new_data
)
SELECT
change_record.lsn,
change_record.xid,
schema_name,
table_name,
operation,
CASE
WHEN operation = ‘DELETE’ THEN
jsonb_build_object(
‘keys’, change_data->’change’->i->’oldkeys’
)
WHEN operation = ‘UPDATE’ THEN
jsonb_build_object(
‘keys’, change_data->’change’->i->’oldkeys’
)
ELSE NULL
END,
CASE
WHEN operation IN (‘INSERT’, ‘UPDATE’) THEN
jsonb_object_agg(
col_name, col_value
)
ELSE NULL
END
FROM
jsonb_array_elements_text(
change_data->’change’->i->’columnnames’
) WITH ORDINALITY AS t(col_name, idx)
JOIN
jsonb_array_elements_text(
change_data->’change’->i->’columnvalues’
) WITH ORDINALITY AS v(col_value, idx)
ON t.idx = v.idx;
END LOOP;
END IF;
END LOOP;
END;
$$ LANGUAGE plpgsql;

— 创建CDC复制槽
SELECT pg_create_logical_replication_slot(‘fgedu_cdc_slot’, ‘wal2json’);

— 测试CDC捕获
INSERT INTO fgedu_orders(order_no, customer_id, amount)
VALUES(‘ORD001’, 1, 1000);

— 输出结果
INSERT 0 1

— 执行CDC捕获
SELECT fgedu_cdc_capture();

— 查看CDC变更记录
SELECT * FROM fgedu_cdc_changes ORDER BY change_time DESC LIMIT 5;

— 输出结果
id | lsn | xid | schema_name | table_name | operation | old_data | new_data | change_time
—-+————+——-+————-+—————+———–+———-+——————————–+—————————-
1 | 0/1A2BC00 | 12352 | public | fgedu_orders | INSERT | NULL | {“order_no”: “ORD001”, …} | 2026-04-07 16:40:00.123456
(1 row)

— 创建同步状态表
CREATE TABLE fgedu_sync_status (
id SERIAL PRIMARY KEY,
source_slot VARCHAR(100),
target_host VARCHAR(100),
target_port INTEGER DEFAULT 5432,
target_db VARCHAR(100),
last_sync_lsn TEXT,
last_sync_time TIMESTAMP,
sync_count BIGINT DEFAULT 0,
status VARCHAR(20) DEFAULT ‘active’
);

— 创建同步日志表
CREATE TABLE fgedu_sync_log (
id SERIAL PRIMARY KEY,
sync_id INTEGER REFERENCES fgedu_sync_status(id),
lsn TEXT,
operation VARCHAR(20),
table_name VARCHAR(100),
record_count INTEGER,
sync_time TIMESTAMP DEFAULT NOW(),
error_message TEXT
);

— 数据同步处理函数
CREATE OR REPLACE FUNCTION fgedu_sync_process()
RETURNS void
AS $$
DECLARE
change_record RECORD;
change_data JSONB;
table_name TEXT;
operation TEXT;
sql_text TEXT;
BEGIN
FOR change_record IN
SELECT lsn, xid, data::jsonb
FROM pg_logical_slot_get_changes(‘fgedu_sync_slot’, NULL, NULL)
LOOP
change_data := change_record.data;

IF change_data ? ‘change’ THEN
FOR i IN 0..jsonb_array_length(change_data->’change’) – 1 LOOP
table_name := change_data->’change’->i->>’table’;
operation := UPPER(change_data->’change’->i->>’kind’);

— 构建同步SQL
CASE operation
WHEN ‘INSERT’ THEN
sql_text := format(
‘INSERT INTO %s (%s) VALUES (%s)’,
table_name,
(
SELECT string_agg(quote_ident(col), ‘,’)
FROM jsonb_array_elements_text(
change_data->’change’->i->’columnnames’
) AS col
),
(
SELECT string_agg(quote_literal(val), ‘,’)
FROM jsonb_array_elements_text(
change_data->’change’->i->’columnvalues’
) AS val
)
);
WHEN ‘UPDATE’ THEN
sql_text := format(
‘UPDATE %s SET %s WHERE %s’,
table_name,
— SET子句
(
SELECT string_agg(
quote_ident(col) || ‘=’ || quote_literal(val),
‘,’
)
FROM
jsonb_array_elements_text(
change_data->’change’->i->’columnnames’
) WITH ORDINALITY AS t(col, idx)
JOIN
jsonb_array_elements_text(
change_data->’change’->i->’columnvalues’
) WITH ORDINALITY AS v(val, idx)
ON t.idx = v.idx
),
— WHERE子句
(
SELECT string_agg(
quote_ident(key) || ‘=’ || quote_literal(val),
‘ AND ‘
)
FROM
jsonb_array_elements_text(
change_data->’change’->i->’oldkeys’->’keynames’
) WITH ORDINALITY AS k(key, idx)
JOIN
jsonb_array_elements_text(
change_data->’change’->i->’oldkeys’->’keyvalues’
) WITH ORDINALITY AS v(val, idx)
ON k.idx = v.idx
)
);
WHEN ‘DELETE’ THEN
sql_text := format(
‘DELETE FROM %s WHERE %s’,
table_name,
(
SELECT string_agg(
quote_ident(key) || ‘=’ || quote_literal(val),
‘ AND ‘
)
FROM
jsonb_array_elements_text(
change_data->’change’->i->’oldkeys’->’keynames’
) WITH ORDINALITY AS k(key, idx)
JOIN
jsonb_array_elements_text(
change_data->’change’->i->’oldkeys’->’keyvalues’
) WITH ORDINALITY AS v(val, idx)
ON k.idx = v.idx
)
);
END CASE;

— 记录同步日志
INSERT INTO fgedu_sync_log(sync_id, lsn, operation, table_name, record_count)
VALUES(1, change_record.lsn, operation, table_name, 1);

— 这里可以执行远程同步
— 使用dblink或外部数据包装器执行sql_text
END LOOP;
END IF;
END LOOP;
END;
$$ LANGUAGE plpgsql;

— 创建同步复制槽
SELECT pg_create_logical_replication_slot(‘fgedu_sync_slot’, ‘wal2json’);

— 测试同步
INSERT INTO fgedu_products(product_name, price, stock)
VALUES(‘Product A’, 99.99, 100);

— 输出结果
INSERT 0 1

— 执行同步处理
SELECT fgedu_sync_process();

— 查看同步日志
SELECT * FROM fgedu_sync_log ORDER BY sync_time DESC LIMIT 5;

— 输出结果
id | sync_id | lsn | operation | table_name | record_count | sync_time
—-+———+————+———–+—————–+————–+————————-
1 | 1 | 0/1A2BD00 | INSERT | fgedu_products | 1 | 2026-04-07 16:45:00.123
(1 row)

— 创建审计日志表
CREATE TABLE fgedu_audit_log (
id SERIAL PRIMARY KEY,
lsn TEXT,
xid BIGINT,
schema_name VARCHAR(100),
table_name VARCHAR(100),
operation VARCHAR(20),
user_name VARCHAR(100),
old_values JSONB,
new_values JSONB,
audit_time TIMESTAMP DEFAULT NOW()
);

CREATE INDEX idx_audit_log_table ON fgedu_audit_log(table_name, audit_time);
CREATE INDEX idx_audit_log_time ON fgedu_audit_log(audit_time);

— 创建审计配置表
CREATE TABLE fgedu_audit_config (
id SERIAL PRIMARY KEY,
table_name VARCHAR(100) NOT NULL,
schema_name VARCHAR(100) DEFAULT ‘public’,
log_insert BOOLEAN DEFAULT TRUE,
log_update BOOLEAN DEFAULT TRUE,
log_delete BOOLEAN DEFAULT TRUE,
enabled BOOLEAN DEFAULT TRUE
);

— 插入审计配置
INSERT INTO fgedu_audit_config(table_name, schema_name)
VALUES
(‘fgedu_accounts’, ‘public’),
(‘fgedu_transactions’, ‘public’);

— 创建审计捕获函数
CREATE OR REPLACE FUNCTION fgedu_audit_capture()
RETURNS void
AS $$
DECLARE
change_record RECORD;
change_data JSONB;
table_name TEXT;
schema_name TEXT;
operation TEXT;
audit_config RECORD;
BEGIN
FOR change_record IN
SELECT lsn, xid, data::jsonb
FROM pg_logical_slot_get_changes(‘fgedu_audit_slot’, NULL, NULL)
LOOP
change_data := change_record.data;

IF change_data ? ‘change’ THEN
FOR i IN 0..jsonb_array_length(change_data->’change’) – 1 LOOP
table_name := change_data->’change’->i->>’table’;
schema_name := change_data->’change’->i->>’schema’;
operation := UPPER(change_data->’change’->i->>’kind’);

— 检查是否需要审计
SELECT * INTO audit_config
FROM fgedu_audit_config
WHERE table_name = table_name
AND schema_name = schema_name
AND enabled = true;

IF FOUND THEN
IF (operation = ‘INSERT’ AND audit_config.log_insert)
OR (operation = ‘UPDATE’ AND audit_config.log_update)
OR (operation = ‘DELETE’ AND audit_config.log_delete) THEN

INSERT INTO fgedu_audit_log(
lsn, xid, schema_name, table_name, operation,
user_name, old_values, new_values
)
VALUES(
change_record.lsn,
change_record.xid,
schema_name,
table_name,
operation,
current_user,
CASE
WHEN operation IN (‘UPDATE’, ‘DELETE’) THEN
change_data->’change’->i->’oldkeys’
ELSE NULL
END,
CASE
WHEN operation IN (‘INSERT’, ‘UPDATE’) THEN
(
SELECT jsonb_object_agg(col, val)
FROM
jsonb_array_elements_text(
change_data->’change’->i->’columnnames’
) WITH ORDINALITY AS t(col, idx)
JOIN
jsonb_array_elements_text(
change_data->’change’->i->’columnvalues’
) WITH ORDINALITY AS v(val, idx)
ON t.idx = v.idx
)
ELSE NULL
END
);
END IF;
END IF;
END LOOP;
END IF;
END LOOP;
END;
$$ LANGUAGE plpgsql;

— 创建审计复制槽
SELECT pg_create_logical_replication_slot(‘fgedu_audit_slot’, ‘wal2json’);

— 测试审计
UPDATE fgedu_accounts SET balance = balance + 100 WHERE account_id = 1;

— 输出结果
UPDATE 1

— 执行审计捕获
SELECT fgedu_audit_capture();

— 查看审计日志
SELECT * FROM fgedu_audit_log ORDER BY audit_time DESC LIMIT 5;

— 输出结果
id | lsn | xid | schema_name | table_name | operation | user_name | old_values | new_values | audit_time
—-+————+——-+————-+—————–+———–+———–+——————————-+———————————-+—————————-
1 | 0/1A2BE00 | 12353 | public | fgedu_accounts | UPDATE | fgedu | {“keynames”: [“account_id”], | {“account_id”: 1, “balance”: | 2026-04-07 16:50:00.123456
| | | | | | | “keyvalues”: [1]} | 1100, …} |
(1 row)