PostgreSQL自定义pgoutput订阅者：流式变更与前置轻量修改问询

针对你的需求，不需要依赖test_decoding的轮询方式，以下几种方案可以实现真正的流式变更接收，并在应用到目标表前完成轻量修改：

方案1：pg_recvlogical + 处理脚本（快速落地） #

pg_recvlogical是PostgreSQL官方自带的命令行工具，支持持续流式拉取逻辑复制槽的变更，无需反复执行SELECT。可以通过管道将变更输出传递给脚本（Python/Shell等），在脚本中完成数据修改后写入目标表。

操作步骤： #

1. 创建基于原生pgoutput插件的逻辑复制槽：

SELECT * FROM pg_create_logical_replication_slot('custom_stream_slot', 'pgoutput');

2. 创建发布（指定要复制的表，这里以全表为例）：

DROP PUBLICATION IF EXISTS custom_pub;
CREATE PUBLICATION custom_pub FOR ALL TABLES;

3. 启动流式接收并对接处理脚本：

pg_recvlogical -d 源数据库名 -S custom_stream_slot -f - --start --option format=json | python change_processor.py

--option format=json让pgoutput输出JSON格式的变更数据，方便脚本解析。

Python处理脚本示例： #

import sys
import json
import psycopg2

# 连接目标数据库
dst_conn = psycopg2.connect("dbname=目标数据库名 user=用户名 password=密码")
dst_cur = dst_conn.cursor()

# 逐行读取流式输出
for line in sys.stdin:
    if not line.strip():
        continue
    change = json.loads(line)
    # 针对INSERT操作做轻量修改（示例：给name字段加后缀）
    if change['change_type'] == 'INSERT':
        change['new_values']['name'] = f"{change['new_values']['name']}_modified"
        # 构造插入语句写入目标表
        col_names = ', '.join(change['new_values'].keys())
        col_values = ', '.join(f"'{v}'" for v in change['new_values'].values())
        dst_cur.execute(f"INSERT INTO {change['table']} ({col_names}) VALUES ({col_values})")
        dst_conn.commit()
    # 可扩展处理UPDATE、DELETE等操作

方案2：基于复制API开发自定义订阅程序（灵活集成） #

如果需要将变更处理逻辑集成到现有应用中，可以直接使用PostgreSQL的复制协议（通过libpq或各语言驱动）实现流式接收，在应用层完成数据修改后写入目标。

Python示例（psycopg2复制接口）： #

import psycopg2
import json
from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT

# 连接源数据库（发布端）
src_conn = psycopg2.connect("dbname=源数据库名 user=用户名")
src_conn.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
src_cur = src_conn.cursor()

# 连接目标数据库
dst_conn = psycopg2.connect("dbname=目标数据库名 user=用户名")
dst_cur = dst_conn.cursor()

# 启动流式复制（使用wal2json插件输出易解析的JSON）
src_cur.execute("START_REPLICATION SLOT custom_stream_slot LOGICAL 0/0 (\"proto_version\" '1', \"publication_names\" 'custom_pub', \"format\" 'json')")

while True:
    msg = src_cur.read_message()
    if not msg:
        continue
    # 解析变更数据
    change_data = json.loads(msg.payload.decode('utf-8'))
    for record in change_data['change']:
        # 修改UPDATE操作的特定字段
        if record['kind'] == 'U':
            record['columnvalues'][1] = f"{record['columnvalues'][1]}_updated"
            # 构造更新语句
            set_clause = ', '.join([f"{col['name']}='{val}'" for col, val in zip(change_data['columns'], record['columnvalues'])])
            dst_cur.execute(f"UPDATE {change_data['table']} SET {set_clause} WHERE id={record['oldkeys']['keyvalues'][0]}")
            dst_conn.commit()
    # 向源库发送反馈，确认已处理该段WAL
    src_cur.send_feedback(flush_lsn=msg.data_start)

注：使用wal2json需要先安装该扩展，创建复制槽时指定'wal2json'作为插件名。

方案3：自定义逻辑解码插件（底层控制） #

如果需要在数据库层面直接处理变更，或者对性能要求极高，可以开发自定义逻辑解码插件（基于C语言），在WAL解析阶段完成数据修改。

核心实现思路： #

• 实现PostgreSQL逻辑解码的核心接口（如decoding_init、decoding_callback）
• 在decoding_callback中解析WAL记录，修改目标字段后再输出变更数据
• 编译安装插件后，创建复制槽时指定该插件即可获取修改后的流式变更

这种方案开发成本较高，但能实现最底层的控制，适合复杂场景。

内容的提问来源于stack exchange，提问作者Stavros Koureas