Python如何自动监听数据库表数据变化并推送到消息队列-Python教程-PHP中文网

必须按数据库类型选择变更监听方案：oracle用cdc，postgresql用listen/notify，mysql解析binlog，sql server走service broker；混用或轮询会导致生产故障。

python如何自动监听数据库表数据变化并推送到消息队列

没有通用的“自动监听”方案，必须根据数据库类型选路：Oracle 用 CDC，PostgreSQL 用 LISTEN/NOTIFY，MySQL 用 binlog，SQL Server 用 Service Broker 或触发器+队列 —— 混用或硬套轮询会卡死生产环境。

Oracle 原生不支持类似 PostgreSQL 的轻量通知机制，cx_Oracle 本身也不提供变更监听接口。靠 time.sleep() 轮询表是反模式：高频率查全表会锁行、拖慢 OLTP；低频又丢变更。

CDC 是 Oracle 官方推荐路径，需 DBA 开启：启用 supplemental logging、创建 capture process、配置 apply process
Python 端用 cx_Oracle 连接时，必须使用 connection.subscribe() 订阅 CDC 变更流（不是普通查询），并设置 qos=ORACLE_SUBSCR_QOS_RELIABLE
若无法开 CDC（如只读账号），退而求其次用物化视图日志（MLOG$）+ 定时查 scn_to_timestamp，但延迟在秒级，且需额外维护日志表

PostgreSQL 的 LISTEN/NOTIFY 是轻量实时机制，Python 通过 psycopg2 可直接绑定，无需中间件。

python-script-generator

快速生成专业的 Python 脚本和应用代码。一键创建完整项目结构，支持CLI、API、爬虫、Bot、Django等多种项目类型，包含完整的项目结构、配置文件、依赖管理、测试、README和文档。

下载

先在数据库建触发器：对目标表的 INSERT/UPDATE/DELETE 执行 NOTIFY table_update, json_build_object(...)
Python 中调用 cursor.execute("LISTEN table_update")，再用 conn.poll() 非阻塞检查，避免线程卡住
注意：NOTIFY 不保证送达，消息不持久；若需可靠投递，必须在收到通知后立刻查变更详情，再发到 RabbitMQ/Kafka
别用 threading.Timer 模拟监听 —— conn.notifies 是队列，要清空（.pop(0)），否则堆积后内存溢出

MySQL 的 binlog 是最接近“真实变更流”的来源，ROW 格式下能精确捕获每行变更，比触发器更可靠、无性能毛刺。

确认 MySQL 已开启 binlog：show variables like 'log_bin';，且 binlog_format = ROW
用 pymysqlreplication 库监听：实例化 BinLogStreamReader，传入 only_events=[WriteRowsEvent, UpdateRowsEvent, DeleteRowsEvent]
触发器方案仅适用于简单场景：触发器写变更 ID 到日志表，Python 轮询该日志表 —— 但高并发下日志表易成瓶颈，且 UPDATE 多字段时难还原完整状态
不要用 SELECT * FROM table 全表比对：10 万行表每次查耗时 200ms+，CPU 和网络压力陡增

SQL Server 触发器不能直接调用外部 HTTP 或 Python 进程，强行用 xp_cmdshell 调脚本是严重安全风险，且不可靠。

标准做法是触发器内执行 SEND ON CONVERSATION 将变更数据发到 Service Broker 队列
Python 用 pyodbc 连接后，执行 WAITFOR (RECEIVE ...) 拉取队列消息，再转投 Kafka/RabbitMQ
若 Service Broker 不可用，改用 SQL Server 的 Change Tracking（非 CDC）：Python 定期查 CHANGETABLE(CHANGES ...)，但需手动维护版本号，延迟约 30 秒起
所有方案都必须处理事务一致性：比如一条 UPDATE 触发多次变更事件，Python 端要按事务 ID 合并，避免重复推送

真正难的不是“怎么推”，而是“怎么确保不丢、不重、不错序”。每个数据库的事务边界、提交时机、错误恢复策略都不同，消息队列端还得配好死信队列和重试间隔 —— 这些细节漏掉一个，上线后就是半夜告警。