Hotdry.
归档
systems-engineering

PostgreSQL WAL 自定义解码插件用于事件源

实现 Postgres WAL 自定义解码插件，支持结构化事件提取、实时事件源管道、模式演进与幂等处理的关键参数与工程实践。

在事件驱动架构中，事件源（Event Sourcing）模式要求将业务变更记录为不可变的事件流，以支持审计、回放和状态重建。PostgreSQL 的预写日志（WAL）提供了一个可靠的变更捕获基础，通过逻辑解码（Logical Decoding）可以将 WAL 中的低级变更转换为高层次的逻辑事件。这不仅避免了应用层触发器的性能瓶颈，还确保了变更的原子性和顺序性。然而，默认的 pgoutput 插件输出二进制格式，难以直接用于事件源的结构化需求，因此实现自定义解码插件是关键扩展点。

自定义 WAL 解码插件的核心在于扩展 PostgreSQL 的输出插件接口（OutputPluginCallbacks），以 pgoutput 为基础，注入事件源特定逻辑，如添加事件元数据（时间戳、用户 ID、聚合根 ID）和处理模式演进。PostgreSQL 文档中描述，逻辑解码从 WAL_level=logical 开始记录足够的信息，支持 DML 变更的解码，但不支持 DDL，这意味着事件源需在应用层管理 schema 变更。根据 Peter Ullrich 的博客，WAL 监听可通过 replication slot 实现实时流式传输，避免了 pg_notify 的队列瓶颈和高吞吐问题。在事件源场景中，这允许将 INSERT/UPDATE/DELETE 映射为事件类型（如 UserCreated、OrderUpdated），并确保事件流的顺序性。

实现自定义插件需使用 C 语言编写动态库（.so 文件），并通过 shared_preload_libraries 加载。插件需实现 startup、begin_txn、change 和 commit_txn 等回调。在 change 回调中，对于 INSERT 操作，可将元组数据序列化为 JSON 事件，包含聚合 ID 以支持 CQRS 分离；对于 UPDATE/DELETE，使用 REPLICA IDENTITY FULL 配置捕获旧值，实现完整事件快照。证据显示，wal2json 等社区插件已证明 JSON 输出在事件源中的可行性，但自定义插件可进一步优化：例如，在输出中嵌入 schema 版本号（如 "schema_version": 2），允许下游消费者处理演进（如添加可选字段，而非破坏性变更）。Peter Ullrich 的 Elixir 实现展示了如何使用 Postgrex.ReplicationConnection 解码 pgoutput 消息，证明了在高并发下的低开销（<1ms 延迟）。

为实现幂等处理，事件需包含唯一标识：使用 WAL 的 LSN（Log Sequence Number）作为事件 ID，或生成 UUID 结合 tx_id（事务 ID）。在下游事件存储（如 Kafka 或专用事件表）中，通过 LSN 去重，避免重复应用。Schema 演进参数包括：事件中可选字段使用 null 填充，版本号控制解析逻辑；回滚策略为事件版本兼容测试，确保旧事件可回放。监控要点：跟踪 WAL 大小（pg_stat_replication_slots.safe_wal_size < 1GB），槽延迟（confirmed_flush_lsn 与当前 LSN 差值 < 10s），插件加载错误（日志中检查 "logical decoding plugin"）。

落地清单：

  1. 配置 postgresql.conf：wal_level=logical, max_replication_slots=10, max_wal_senders=20，重启生效。
  2. 创建 publication：CREATE PUBLICATION event_pub FOR ALL TABLES WITH (publish='insert,update,delete')；
  3. 创建 slot：SELECT pg_create_logical_replication_slot ('event_slot', 'custom_plugin')；
  4. 插件代码框架（C）： 
    #include "postgres.h"
#include "replication/output_plugin.h"

void startup(LogicalDecodingContext *ctx, OutputPluginOptions *opt, bool is_init) {
    // 初始化事件版本
    opt->output_json = true;
}

void change(LogicalDecodingContext *ctx, ReorderBufferTXN *txn, Relation rel, ReorderBufferChange *change) {
    // 提取元组，序列化为 JSON 事件，添加 LSN 和 schema_version
    OutputPluginPrepareWrite(ctx, true);
    appendStringInfo(ctx->out, "{\"type\":\"%s\",\"data\":%s,\"lsn\":\"%s\",\"version\":2}", change_kind, json_data, lsn_str);
    OutputPluginWrite(ctx, false);
}
    编译：gcc -shared -o custom_plugin.so custom_plugin.c -I$PG_INCLUDEDIR ...
  5. 加载：shared_preload_libraries='custom_plugin'。
  6. Elixir 消费者：使用 Postgrex.ReplicationConnection 启动流，解码 JSON 事件，推送到事件总线。
  7. 测试：插入数据，验证事件流顺序和幂等（模拟重启，检查去重）。

此方案在生产中可扩展到多租户，通过 publication 过滤表，支持 1M+ TPS 变更捕获。风险包括 WAL 膨胀（定期清理槽）和 C 代码维护，但通过 Docker 镜像分发可缓解。

资料来源：

查看归档