你有没有想过，数据库里每一条 INSERT、UPDATE、DELETE，能不能实时抓到。
不是事后翻 binlog，不是等 T+1 的离线同步。是那种一条变更刚提交，你这边就收到了的感觉。
我说的是，用 Python 直接消费数据库的 WAL 流。
我以前做过一个项目，甲方要求「任何人对核心表的数据变更，30 秒内必须同步到审计系统」。30 秒。不是 30 分钟。当时第一个反应是，这得直接读日志了吧。
后来在金仓上跑通了。decoderbufs 插件 + 一段 Python 脚本，搞定。
今天跟你聊聊这事怎么做。
WAL，Write-Ahead Logging，说白了就是数据库的「先写日志再落盘」机制。
每次你执行一条 INSERT，数据库不是直接往数据文件里写的。它先把这条变更记到 WAL 日志里，然后才慢慢把数据刷到磁盘。
这条日志就像数据库的黑匣子。不管数据库是正常关的还是被人拔了电源，只要 WAL 在，数据就能恢复。
但问题是，原始的 WAL 是二进制格式的。你看一眼，满屏的十六进制，跟天书一样。
00 1A 2F 3C 00 00 00 05 49 4E 53 45 52 54 00 00
这不是给人看的东西。这是给数据库自己恢复用的。
专业术语叫「物理 WAL」，它记录的是哪个数据块的哪个字节偏移量被改成了什么值。你知道有变更发生了，但你完全看不懂到底改了啥。
物理 WAL 是保安的监控录像，逻辑 WAL 才是值班日志
你需要的是「逻辑解码」。
什么意思呢。物理 WAL 像是一段没有标注的监控录像。画面里有人进出，但你不知道是谁，从哪来，到哪去。逻辑解码就是给监控录像加上人脸识别和时间戳，告诉你「张三，14 点 32 分，从表 users 里改了一行，把 email 从 old@123.com 改成了 new@456.com」。
往技术层面说，逻辑解码把 WAL 里那些二进制字节，解析成结构化的数据。表名、操作类型、字段名、旧值、新值，一目了然。
金仓的逻辑解码靠的是 decoderbufs 插件。这插件是金仓自带的，不用额外装。
它的工作原理不复杂。数据库创建一个「复制槽」，decoderbufs 插件把这个槽里的 WAL 变更转成 Protocol Buffers 格式的数据流。Protobuf 是一种紧凑的结构化数据格式，比 JSON 体积小，解析快。
你的 Python 程序连到这个复制槽，就能源源不断地接收到结构化的 DML 变更数据。
流程图大概长这样。
你的 INSERT/UPDATE/DELETE
↓
数据库 WAL 日志
↓
decoderbufs 插件（逻辑解码）
↓
复制槽（replication slot）→ Protobuf 数据流
↓
Python 脚本（sys_recvlogical → 解析 → 输出）
说干就干。
第一步，改 kingbase.conf。几个关键参数得打开。
wal_level = logical
max_replication_slots = 4
max_wal_senders = 4
wal_level 设成 logical，数据库才会在 WAL 里记录足够的逻辑信息。光有物理日志是不够做解码的。另外两个参数控制允许同时建几个复制槽、起几个 WAL 发送进程。4 个一般在中小规模场景下够用了。
改完重启数据库。这个没捷径，必须重启。
第二步，给权限。执行复制的用户需要 REPLICATION 权限。
ALTER USER kingbase_user REPLICATION;
还有 sys_hba.conf 里要加一条复制连接的规则。
host replication kingbase_user 0.0.0.0/0 md5
第三步，也是最容易被坑的，编码。
环境是 KingbaseES V009R001C002B0014，Oracle 模式。如果你在建库的时候没注意字符集，数据里混了 GBK、Latin1，后面解析的时候中文全是乱码。
先确认编码是不是 UTF8。
SHOW server_encoding;
SHOW client_encoding;
两条都必须是 UTF8。不是的话，改表的字符集。
ALTER TABLE 表名 ALTER COLUMN 字段名 TYPE varchar(255) CHARACTER SET utf8;
数据库配好了，接下来建复制槽。
SELECT sys_create_logical_replication_slot('wal_decoder_slot', 'decoderbufs');
这个 slot 就是你和 WAL 流之间的管道。第二个参数 'decoderbufs' 指定用哪个解码插件。
slot 建好之后，数据库就会一直给你保留 WAL 日志，直到你消费掉。你停了的这段时间，WAL 在磁盘上攒着，不丢数据。
但换个角度看，你如果停了很久不消费，WAL 会越攒越多，磁盘满了数据库就挂了。所以解析停掉之后，一定记得清理。
SELECT sys_drop_replication_slot('wal_decoder_slot');
这事儿我踩过坑。测试环境跑了两天忘了停，WAL 攒了 80 多个 GB，数据库直接拒绝写入。不是技术难，是你根本想不到会是这个原因。
终于到正餐了。
消费 WAL 流的核心工具是 sys_recvlogical。这是金仓自带的逻辑复制客户端，类比 PostgreSQL 的 pg_recvlogical。它能连接到复制槽，把解码后的数据流源源不断地吐出来。
Python 这边做三件事。启动 sys_recvlogical 并读取它的输出流。把收到的 Protobuf 数据解析成可读的字段。处理编码问题防止中文乱码。
依赖先装好。
pip install protobuf chardet
还有一个容易被忽视的步骤。decoderbufs 输出的 Protobuf 数据需要 decoderbufs_pb2.py 这个解析文件。这个文件怎么来的。从金仓安装目录找到 decoderbufs.proto，用 protoc 编译。
protoc --python_out=. decoderbufs.proto
编译完会生成 decoderbufs_pb2.py，放到你脚本的同目录下，import 就行。
完整脚本长这样。
import os
import subprocess
import sys
import chardet
# 强制客户端 UTF8 编码
os.environ["PGCLIENTENCODING"] = "UTF8"
# 兜底编码检测
def detect_encoding(raw_bytes):
result = chardet.detect(raw_bytes)
encoding = result.get("encoding", "utf-8")
if encoding is None:
encoding = "utf-8"
try:
return raw_bytes.decode(encoding)
except (UnicodeDecodeError, LookupError):
return raw_bytes.decode("utf-8", errors="replace")
# 导入由 decoderbufs.proto 编译生成的 Python 模块
try:
import decoderbufs_pb2
except ImportError:
print("找不到 decoderbufs_pb2.py，请先编译 decoderbufs.proto")
sys.exit(1)
def consume_wal(slot_name="wal_decoder_slot"):
cmd = [
"sys_recvlogical",
"--slot", slot_name,
"--start",
"-f", "-",
"-d", "kingbase",
"-U", "kingbase_user",
"-W",
"-o", "include-timestamp=true"
]
proc = subprocess.Popen(
cmd,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
print(f"复制槽 {slot_name} 已连接，等待 DML 变更...")
try:
while True:
raw = proc.stdout.readline()
if not raw:
continue
msg = decoderbufs_pb2.RowMessage()
msg.ParseFromString(raw)
op_map = {"INSERT": "插入", "UPDATE": "更新", "DELETE": "删除"}
op = op_map.get(str(msg.op), str(msg.op))
table = msg.table
ts = msg.timestamp
print(f"[{ts}] {op} | 表 {table}")
for col_data in msg.new_tuple:
col_name = col_data.column_name
col_value = detect_encoding(col_data.datum)
print(f" {col_name} → {col_value}")
print("-" * 50)
except KeyboardInterrupt:
print("手动停止")
finally:
proc.terminate()
if __name__ == "__main__":
consume_wal()
脚本不长，但几个地方值得说一下。
os.environ["PGCLIENTENCODING"] = "UTF8" 这一行强制客户端编码跟服务端保持一致。我见过太多人卡在「为什么中文全是问号」这一步。
detect_encoding 函数是兜底。chardet 自动检测实际编码，能转换就转，不能转就替换成占位符。宁可丢一个字，也别让整个解析流程崩掉。
还有就是 Python 脚本和 decoderbufs_pb2.py 要在一个目录下，不然 import 失败。
脚本启动后，你会看到这行提示。
复制槽 wal_decoder_slot 已连接，等待 DML 变更...
然后去数据库里随便跑几条 DML，含中文的那种。
INSERT INTO test_order (order_id, customer, amount) VALUES (10001, '张三', 12580.00);
UPDATE test_order SET amount = 15800.00 WHERE order_id = 10001;
DELETE FROM test_order WHERE order_id = 10001;
Python 这边马上就会有输出。
[2026-04-20 14:32:01] 插入 | 表 test_order
order_id → 10001
customer → 张三
amount → 12580.00
--------------------------------------------------
[2026-04-20 14:32:15] 更新 | 表 test_order
order_id → 10001
customer → 张三
amount → 15800.00
--------------------------------------------------
[2026-04-20 14:32:28] 删除 | 表 test_order
order_id → 10001
customer → 张三
amount → 15800.00
--------------------------------------------------
中文正常显示，字段值一一对应。从 INSERT 执行到 Python 输出，延迟基本在 1 秒以内。
搞了半个小时，终于通了。那种感觉怎么说呢，就像你把耳朵贴到数据库引擎上，听到了每一次数据呼吸的声音。
实时数据同步。你有一个 OLTP 数据库和一个 OLAP 查询库，想让查询库的数据尽量实时。在 OLTP 上起一个 slot，Python 消费 WAL 变更，实时写入 OLAP 库。比定时批量同步快得多。
审计日志。谁在什么时候改了哪张表的哪个字段，旧值是什么，新值是什么，全部记录在案。不是应用层的日志，是数据库内核级别的审计，绕过不了。
实时数据管道。把你数据库里的变更加上一些业务逻辑，推到 Kafka、Elasticsearch、Redis 里去。下游系统自己消费，不用反复查库。
数据迁移。OLTP 切 OLAP 或异构库的时候，先用快照同步存量，再用 WAL 消费追增量。切换的时候数据差距最小。
protobuf 文件版本要对齐。必须用金仓安装目录自带的 decoderbufs.proto 来编译。网上下载的 PostgreSQL 版本的 proto 文件可能不兼容。字段定义不一样，解析直接报错。一个字节都不差，这是铁律。
复制槽是双刃剑。好处是不丢数据，坏处是一直占着 WAL。假如你的 Python 脚本挂了，三天没重启，WAL 攒到磁盘满。数据库拒绝写入的那一刻，全业务挂掉。所以生产上要么加监控，要么把 slot 的生命周期管理好。
编码是隐形杀手。很多项目的数据库不是纯 UTF8 环境，历史数据里可能混了 GBK 甚至 Latin1。Python 脚本不加 chardet 兜底，遇到一条非 UTF8 的数据就崩了。别猜，别大概，提前确认字符集。
WAL 流量比你想象的大。一个高频写入的生产库，WAL 可能跑满你的网络带宽。解码本身也吃 CPU。如果是大并发的 OLTP 场景，先在测试环境评估一下吞吐量，别直接往生产上挂。
以上，既然看到这里了，如果觉得不错，随手点个赞、在看、转发三连吧，如果想第一时间收到推送，也可以给我个星标。
谢谢你看我的文章，我们，下次再见。
转发给那个正在搞实时数据同步的同事，他知道我在说什么。
