本文目录导读：
针对Python脚本适配实时触发同步场景，主要有以下几种方案,按推荐程度排序：
轮询 + 状态标记法（最简单）
import time
import hashlib
import os
class FileSyncTrigger:
def __init__(self, watch_dir, sync_interval=5):
self.watch_dir = watch_dir
self.sync_interval = sync_interval
self.last_checksum = {}
def get_file_hash(self, filepath):
"""计算文件MD5用于检测变化"""
with open(filepath, 'rb') as f:
return hashlib.md5(f.read()).hexdigest()
def monitor_and_sync(self):
"""主循环：轮询检测变化"""
while True:
for root, dirs, files in os.walk(self.watch_dir):
for file in files:
filepath = os.path.join(root, file)
current_hash = self.get_file_hash(filepath)
if current_hash != self.last_checksum.get(filepath):
print(f"[变更检测] {filepath}")
self.trigger_sync(filepath)
self.last_checksum[filepath] = current_hash
time.sleep(self.sync_interval)
def trigger_sync(self, filepath):
"""触发同步逻辑 - 可替换为实际同步操作"""
print(f"执行同步: {filepath}")
# 在这里实现你的同步代码
# 使用示例
sync = FileSyncTrigger("/path/to/sync/folder")
sync.monitor_and_sync()
Watchdog库（事件驱动，推荐）
import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class SyncHandler(FileSystemEventHandler):
def on_modified(self, event):
"""文件被修改时触发"""
if not event.is_directory:
print(f"文件修改: {event.src_path}")
self.execute_sync(event.src_path)
def on_created(self, event):
"""文件被创建时触发"""
if not event.is_directory:
print(f"文件创建: {event.src_path}")
self.execute_sync(event.src_path)
def on_deleted(self, event):
"""文件被删除时触发"""
print(f"文件删除: {event.src_path}")
# 删除类同步可以在这里处理
def execute_sync(self, filepath):
"""实际的同步逻辑"""
# 可以添加延迟去重，避免高频触发
print(f"开始同步: {filepath}")
# 在这里实现你的同步代码
class RealtimeSync:
def __init__(self, watch_paths):
self.paths = watch_paths if isinstance(watch_paths, list) else [watch_paths]
self.observer = Observer()
def start(self):
"""启动监控"""
for path in self.paths:
handler = SyncHandler()
self.observer.schedule(handler, path, recursive=True)
self.observer.start()
print(f"监控已启动: {self.paths}")
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
self.observer.stop()
self.observer.join()
def stop(self):
"""停止监控"""
self.observer.stop()
# 使用示例
sync = RealtimeSync(["/path/to/sync/folder1", "/path/to/sync/folder2"])
sync.start()
异步+回调模式（高性能场景）
import asyncio
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from concurrent.futures import ThreadPoolExecutor
class AsyncSyncHandler(FileSystemEventHandler):
def __init__(self, async_callback):
self.async_callback = async_callback
self.executor = ThreadPoolExecutor(max_workers=4)
def on_modified(self, event):
if not event.is_directory:
asyncio.run(self.handle_event_async(event.src_path))
async def handle_event_async(self, filepath):
"""异步处理同步事件"""
print(f"接收到变更: {filepath}")
# 并行执行多个同步任务
tasks = [
self.sync_to_remote(filepath),
self.update_cache(filepath),
self.log_event(filepath)
]
await asyncio.gather(*tasks)
async def sync_to_remote(self, filepath):
"""模拟远程同步"""
await asyncio.sleep(0.5) # 模拟网络IO
print(f"远程同步完成: {filepath}")
async def update_cache(self, filepath):
"""更新本地缓存"""
await asyncio.sleep(0.1)
print(f"缓存更新: {filepath}")
async def log_event(self, filepath):
"""日志记录"""
print(f"日志记录: {filepath}")
class AsyncRealtimeSync:
def __init__(self, watch_path):
self.observer = Observer()
self.handler = AsyncSyncHandler(self.on_event)
def on_event(self, filepath):
"""事件回调"""
pass
def start(self):
self.observer.schedule(self.handler, self.watch_path, recursive=True)
self.observer.start()
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_forever()
# 使用示例
sync = AsyncRealtimeSync("/path/to/sync")
sync.start()
结合消息队列（分布式场景）
import pika
import json
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class MessageQueueSync(FileSystemEventHandler):
def __init__(self, rabbitmq_host='localhost'):
self.connection = pika.BlockingConnection(
pika.ConnectionParameters(rabbitmq_host)
)
self.channel = self.connection.channel()
self.channel.queue_declare(queue='sync_events')
def on_modified(self, event):
if not event.is_directory:
message = {
'action': 'sync',
'path': event.src_path,
'type': 'modified'
}
self.channel.basic_publish(
exchange='',
routing_key='sync_events',
body=json.dumps(message)
)
print(f"已发送同步消息: {event.src_path}")
# 消费者示例
def sync_consumer():
connection = pika.BlockingConnection(
pika.ConnectionParameters('localhost')
)
channel = connection.channel()
channel.queue_declare(queue='sync_events')
def callback(ch, method, properties, body):
event = json.loads(body)
print(f"收到同步任务: {event}")
# 执行实际的同步逻辑
channel.basic_consume(
queue='sync_events',
on_message_callback=callback,
auto_ack=True
)
channel.start_consuming()
性能优化技巧
防抖处理（高频避免）
import time
from threading import Timer
class DebounceSync:
def __init__(self, delay=0.5):
self.delay = delay
self.timer = None
self.last_event_time = 0
def on_file_changed(self, filepath):
current_time = time.time()
# 防止短时间内重复触发
if current_time - self.last_event_time < 0.1:
return
self.last_event_time = current_time
# 重置定时器 - 延迟执行，合并连续变更
if self.timer:
self.timer.cancel()
self.timer = Timer(self.delay, self.execute_sync, args=[filepath])
self.timer.start()
def execute_sync(self, filepath):
print(f"最终执行同步: {filepath}")
选择建议
| 场景 | 推荐方案 |
|---|---|
| 简单文件监控 | Watchdog库 |
| 高性能要求 | 异步 + Watchdog |
| 分布式系统 | 消息队列方案 |
| 快速原型 | 轮询方案 |
| 低延迟要求 | inotify（Linux原生） |
最佳实践：
- 使用Watchdog库作为基础
- 添加防抖机制应对高频写入
- 使用异步IO处理网络同步
- 加入错误重试和日志记录
- 考虑使用缓存减少重复计算
