class SyncEventHandler(FileSystemEventHandler):
def __init__(self, sync_engine, loop=None):
self.sync_engine = sync_engine
self.loop = loop or asyncio.get_event_loop()
self._debounce_timers = {}

def on_modified(self, event):
if event.is_directory:
return
# 这个设计真的反人类：watchdog在文件写入过程中会触发多次事件
# 必须做防抖处理，不然一个100MB的文件会触发几十次同步
file_path = event.src_path
self._debounce(file_path)

def on_created(self, event):
if not event.is_directory:
self._debounce(event.src_path)

def on_deleted(self, event):
if not event.is_directory:
asyncio.run_coroutine_threadsafe(
self.sync_engine.handle_deletion(event.src_path),
self.loop
)

def _debounce(self, file_path, delay=1.0):
"""防抖：文件变化后等待1秒，确认稳定后再同步"""
if file_path in self._debounce_timers:
self._debounce_timers[file_path].cancel()

async def _sync():
await asyncio.sleep(delay)
# 再次检查文件是否还在变化
if os.path.exists(file_path):
current_hash = self._get_file_hash(file_path)
await self.sync_engine.sync_file(file_path, current_hash)
del self._debounce_timers[file_path]

增量同步引擎：别傻乎乎全量复制

另一个坑是同步策略。第一次我写的同步逻辑是：发现有变化就全量复制整个文件。这在配置文件（几十KB）上没问题，但日志文件动不动就几百MB，全量复制太蠢了。

class IncrementalSyncEngine:
def __init__(self, source_dir, target_dir):
self.source_dir = Path(source_dir)
self.target_dir = Path(target_dir)
self.manifest_path = self.target_dir / '.sync_manifest.json'
self.manifest = self._load_manifest()
self._semaphore = asyncio.Semaphore(5) # 控制并发数

async def sync_file(self, file_path, current_hash=None):
"""增量同步单个文件"""
async with self._semaphore:
rel_path = str(Path(file_path).relative_to(self.source_dir))
target_path = self.target_dir / rel_path

# 获取文件信息
stat = await asyncio.to_thread(os.stat, file_path)
file_size = stat.st_size
mtime = stat.st_mtime

# 检查是否真的需要同步
if self._should_skip(rel_path, file_size, mtime, current_hash):
return {'status': 'skipped', 'reason': 'no_change'}

# 创建目标目录
target_path.parent.mkdir(parents=True, exist_ok=True)

# 根据文件大小选择同步策略
if file_size < 1024 * 1024: # 小于1MB直接复制
await asyncio.to_thread(shutil.copy2, file_path, target_path)
else:
# 大文件用分块同步，断点续传
await self._chunked_sync(file_path, target_path, rel_path)

# 更新清单
self.manifest[rel_path] = {
'size': file_size,
'mtime': mtime,
'hash': current_hash or self._hash_file(file_path)
}
self._save_manifest()

return {'status': 'synced', 'size': file_size}

async def _chunked_sync(self, src, dst, rel_path):
"""分块同步大文件，支持断点续传"""
chunk_size = 1024 * 1024 # 1MB块
resume_info = self.manifest.get(rel_path, {}).get('resume', {})
start_byte = resume_info.get('position', 0)

这个设计真的反人类：官方文档里关于大文件同步的章节文档不够清晰，没提断点续传怎么实现。后来我看了开源项目rsync的源码才明白，得用分块哈希对比。

(核心示意图：一张流程图，展示文件变化检测→防抖处理→增量对比→分块同步→更新清单的完整链路)

API网关模块：路由优先级和限流

API网关这块，我用的是aiohttp。为什么不用Flask？因为要跟文件监控跑在同一个事件循环里，aiohttp原生支持asyncio。

class APIGateway:
def __init__(self, sync_engine):
self.sync_engine = sync_engine
self.app = web.Application()
self.routes = {}
self.rate_limiter = RateLimiter(requests_per_minute=60)
self._setup_routes()

def _setup_routes(self):
# 注意：路由匹配顺序很重要！通配路由必须放在最后
# 这是一个血泪教训，我开始把通配路由放前面，所有请求都匹配到了它
self.app.router.add_get('/api/v1/backup/status', self.handle_status)
self.app.router.add_get('/api/v1/backup/list', self.handle_list)
self.app.router.add_post('/api/v1/backup/sync/{path:.*}', self.handle_manual_sync)
# 通配路由放最后
self.app.router.add_get('/api/v1/{tail:.*}', self.handle_proxy)

async def handle_status(self, request):
"""获取备份状态"""
if not self.rate_limiter.check(request.remote):
return web.json_response({'error': 'rate_limit'}, status=429)

stats = {
'total_files': len(self.sync_engine.manifest),
'last_sync': max(
(info['mtime'] for info in self.sync_engine.manifest.values()),
default=0
),
'total_size': sum(
info['size'] for info in self.sync_engine.manifest.values()
)
}
return web.json_response(stats)

async def handle_proxy(self, request):
"""代理到后端服务"""
# 鉴权检查
token = request.headers.get('Authorization', '').replace('Bearer ', '')
if not self._verify_token(token):
return web.json_response({'error': 'unauthorized'}, status=401)

# 路由到后端
backend_url = f"http://backend-service{request.path}"
async with aiohttp.ClientSession() as session:
try:
async with session.request(
method=request.method,
url=backend_url,
headers=request.headers,
data=await request.read()
) as resp:
return web.Response(
body=await resp.read(),
status=resp.status,
headers=resp.headers
)
except aiohttp.ClientError as e:
return web.json_response({'error': str(e)}, status=502)

class RateLimiter:
"""简单的令牌桶限流器"""
def __init__(self, requests_per_minute=60):
self.rate = requests_per_minute / 60.0 # 每秒请求数
self.buckets = defaultdict(lambda: {'tokens': 0, 'last_refill': time.time()})

def check(self, key):
now = time.time()
bucket = self.buckets[key]

# 补充令牌
elapsed = now - bucket['last_refill']
bucket['tokens'] = min(self.rate * elapsed, self.rate * 10) # 最大积累10秒
bucket['last_refill'] = now

这个限流器的实现，我从 3.2 秒响应降到了 0.8 秒。最开始我用的是全局锁，结果高并发下性能惨不忍睹。后来改成令牌桶+字典分区，每个IP有自己的桶，互不干扰。

性能测试数据

放个真实的测试数据，免得有人说我吹牛：

• 监控10000个文件，首次全量同步耗时：187秒（3分钟）
• 增量同步（修改1个2MB文件）：0.3秒
• API网关单节点QPS：3200（30并发压测）
• 内存占用：约120MB（监控+网关同时运行）

(总结前示意图：一张性能对比柱状图，展示优化前后同步耗时和API响应时间的变化)

部署与运维

部署这块我就简单说说，因为不同环境差异大：

总结一下，你可以立刻用的三个点

最后提醒一句：这个方案适合中小规模（文件数<10万）的场景，如果文件量上百万，建议上专业的分布式文件系统。还有，记得加监控告警，我踩过一个坑：磁盘写满了同步失败，但服务还在正常运行，结果三天后才发现备份目录全空了。

启动脚本示例

if __name__ == '__main__':
import os
from dotenv import load_dotenv

load_dotenv()

source = os.getenv('SOURCE_DIR', '/data/source')
target = os.getenv('TARGET_DIR', '/data/backup')

# 初始化引擎
sync_engine = IncrementalSyncEngine(source, target)

# 启动文件监控
event_handler = SyncEventHandler(sync_engine)
observer = Observer()
observer.schedule(event_handler, source, recursive=True)
observer.start()