Python脚本如何实时感知业务数据变更

本文目录导读：

1. 数据库监听（最常用）
2. 消息队列监听
3. 文件系统监听（针对文件存储）
4. Redis发布/订阅
5. 定时轮询（简单但非实时）
6. WebSocket实时推送
7. 最佳实践建议

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())

最佳实践建议

1. 根据场景选择方案：

   • 数据库监听：适合需要实时同步数据库变更
   • 消息队列：适合分布式系统解耦
   • WebSocket：适合前端实时展示

2. 考虑性能影响：

   • 避免频繁轮询
   • 设置合理的超时时间
   • 使用连接池管理数据库连接

3. 错误处理：

   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

4. 资源清理：

   try:
       # 监听逻辑
       pass
   finally:
       # 确保资源释放
       if conn:
           conn.close()

根据你的具体业务场景（数据存储方式、实时性要求、系统架构等），选择合适的方案，如果需要更具体的指导，请提供更多背景信息。