本文目录导读：
Python脚本要实现实时感知业务数据变更,通常需要根据数据存储方式选择不同的方案，以下是几种主流且实用的方法：
数据库监听（最常用）
PostgreSQL - LISTEN/NOTIFY
import psycopg2
import select
def listen_to_changes():
conn = psycopg2.connect("dbname=mydb user=myuser password=mypass")
conn.set_isolation_level(psycopg2.extensions.ISOLATION_LEVEL_AUTOCOMMIT)
cur = conn.cursor()
cur.execute("LISTEN my_channel;")
print("Waiting for notifications...")
while True:
if select.select([conn], [], [], 5) != ([], [], []):
conn.poll()
while conn.notifies:
notify = conn.notifies.pop(0)
print(f"Got notification: {notify.payload}")
# 处理变更的业务逻辑
process_change(notify.payload)
MySQL - binlog监听
from pymysqlreplication import BinLogStreamReader
from pymysqlreplication.row_event import (
WriteRowsEvent, UpdateRowsEvent, DeleteRowsEvent
)
def listen_mysql_binlog():
stream = BinLogStreamReader(
connection_settings={
"host": "localhost",
"port": 3306,
"user": "replicator",
"passwd": "password"
},
server_id=100,
only_events=[WriteRowsEvent, UpdateRowsEvent, DeleteRowsEvent],
only_tables=["your_table"]
)
for event in stream:
if isinstance(event, WriteRowsEvent):
for row in event.rows:
print(f"New row: {row['values']}")
elif isinstance(event, UpdateRowsEvent):
for row in event.rows:
print(f"Updated: {row['before_values']} -> {row['after_values']}")
elif isinstance(event, DeleteRowsEvent):
for row in event.rows:
print(f"Deleted: {row['values']}")
消息队列监听
import pika
import json
def callback(ch, method, properties, body):
data = json.loads(body)
print(f"Received data change: {data}")
# 处理业务逻辑
process_business_logic(data)
# RabbitMQ示例
connection = pika.BlockingConnection(
pika.ConnectionParameters('localhost')
)
channel = connection.channel()
channel.queue_declare(queue='data_changes')
channel.basic_consume(
queue='data_changes',
on_message_callback=callback,
auto_ack=True
)
print('Waiting for data changes...')
channel.start_consuming()
文件系统监听（针对文件存储）
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class DataChangeHandler(FileSystemEventHandler):
def on_modified(self, event):
if not event.is_directory:
print(f"File changed: {event.src_path}")
# 读取变更后的数据
with open(event.src_path, 'r') as f:
data = f.read()
process_data_change(data)
def on_created(self, event):
print(f"New file created: {event.src_path}")
# 启动监听
observer = Observer()
handler = DataChangeHandler()
observer.schedule(handler, path='/path/to/watch', recursive=True)
observer.start()
Redis发布/订阅
import redis
def listen_redis_changes():
r = redis.Redis(host='localhost', port=6379, db=0)
pubsub = r.pubsub()
pubsub.subscribe('data_channel')
for message in pubsub.listen():
if message['type'] == 'message':
data = message['data'].decode('utf-8')
print(f"Received: {data}")
# 处理变更
process_changed_data(data)
定时轮询（简单但非实时）
import time
import hashlib
from sqlalchemy import create_engine, text
class DataPoller:
def __init__(self, interval=5):
self.interval = interval
self.engine = create_engine('postgresql://user:pass@localhost/db')
self.last_checksum = None
def get_data_checksum(self):
with self.engine.connect() as conn:
result = conn.execute(text("SELECT MD5(array_agg(id || '-' || version)::text) FROM your_table"))
return result.fetchone()[0]
def start_polling(self):
while True:
current_checksum = self.get_data_checksum()
if current_checksum != self.last_checksum:
print("Data changed!")
self.last_checksum = current_checksum
# 执行变更处理
self.handle_data_change()
time.sleep(self.interval)
WebSocket实时推送
import asyncio
import websockets
import json
async def listen_websocket():
async with websockets.connect('ws://your-server/events') as websocket:
while True:
message = await websocket.recv()
data = json.loads(message)
print(f"Received real-time update: {data}")
# 处理业务逻辑
process_update(data)
# 运行
asyncio.run(listen_websocket())
最佳实践建议
-
根据场景选择方案：
- 数据库监听：适合需要实时同步数据库变更
- 消息队列：适合分布式系统解耦
- WebSocket：适合前端实时展示
-
考虑性能影响：
- 避免频繁轮询
- 设置合理的超时时间
- 使用连接池管理数据库连接
-
错误处理：
def safe_listener(): while True: try: listen_to_changes() except Exception as e: print(f"Error: {e}, reconnecting in 5 seconds...") time.sleep(5) continue -
资源清理：
try: # 监听逻辑 pass finally: # 确保资源释放 if conn: conn.close()
根据你的具体业务场景（数据存储方式、实时性要求、系统架构等），选择合适的方案，如果需要更具体的指导，请提供更多背景信息。
