定时批量合并（基础方案）
import time
import threading
from collections import deque
from datetime import datetime, timedelta
class BatchSyncTask:
def __init__(self, batch_interval=5, max_batch_size=10):
self.task_queue = deque()
self.batch_interval = batch_interval # 批量间隔（秒）
self.max_batch_size = max_batch_size # 最大批量大小
self.lock = threading.Lock()
self.timer = None
def add_task(self, task_data):
"""添加单个任务到队列"""
with self.lock:
self.task_queue.append({
'data': task_data,
'timestamp': datetime.now()
})
# 如果达到最大批量，立即处理
if len(self.task_queue) >= self.max_batch_size:
self._process_batch()
# 否则启动定时器
elif self.timer is None:
self._start_timer()
def _start_timer(self):
"""启动定时器"""
self.timer = threading.Timer(self.batch_interval, self._process_batch)
self.timer.start()
def _process_batch(self):
"""处理批量任务"""
with self.lock:
if not self.task_queue:
return
# 获取当前批次的所有任务
batch = list(self.task_queue)
self.task_queue.clear()
self.timer = None
# 执行批量同步
self._sync_batch(batch)
def _sync_batch(self, batch):
"""实际的批量同步逻辑"""
print(f"同步 {len(batch)} 个任务: {[t['data'] for t in batch]}")
# 这里实现具体的同步逻辑
# 使用示例
batch_sync = BatchSyncTask(batch_interval=5, max_batch_size=10)
# 模拟低频任务
for i in range(5):
batch_sync.add_task(f"task_{i}")
time.sleep(1) # 每隔1秒添加一个任务
异步 + 回调模式
import asyncio
from asyncio import Queue
import time
class AsyncBatchSync:
def __init__(self, batch_interval=5, max_batch_size=10):
self.queue = Queue()
self.batch_interval = batch_interval
self.max_batch_size = max_batch_size
self.processing = False
async def add_task(self, task_data):
"""异步添加任务"""
await self.queue.put(task_data)
# 启动处理协程（如果未运行）
if not self.processing:
asyncio.create_task(self._process_loop())
async def _process_loop(self):
"""批量处理循环"""
self.processing = True
batch = []
last_process_time = time.time()
while True:
try:
# 等待新任务（带超时）
task = await asyncio.wait_for(
self.queue.get(),
timeout=self.batch_interval
)
batch.append(task)
# 检查是否达到批量大小或时间间隔
current_time = time.time()
if (len(batch) >= self.max_batch_size or
current_time - last_process_time >= self.batch_interval):
await self._sync_batch(batch)
batch = []
last_process_time = current_time
except asyncio.TimeoutError:
# 超时：处理剩余任务
if batch:
await self._sync_batch(batch)
batch = []
last_process_time = time.time()
# 检查是否还有更多任务
if self.queue.empty():
self.processing = False
break
async def _sync_batch(self, batch):
"""执行批量同步"""
print(f"异步批量同步 {len(batch)} 个任务: {batch}")
# 模拟异步操作
await asyncio.sleep(1)
# 实际同步逻辑...
# 使用示例（需要异步上下文）
async def main():
sync = AsyncBatchSync(batch_interval=3)
# 模拟低频任务
for i in range(6):
await sync.add_task(f"async_task_{i}")
await asyncio.sleep(0.5) # 每0.5秒添加一个
# asyncio.run(main())
装饰器方法
from functools import wraps
import threading
import time
def batch_sync(batch_interval=5, max_batch_size=10):
"""批量同步装饰器"""
def decorator(func):
# 存储每个函数的队列和锁
queues = {}
locks = {}
timers = {}
@wraps(func)
def wrapper(*args, **kwargs):
# 使用函数名作为唯一标识
func_id = func.__name__
# 初始化队列和锁
if func_id not in queues:
queues[func_id] = []
locks[func_id] = threading.Lock()
timers[func_id] = None
with locks[func_id]:
queues[func_id].append((args, kwargs, time.time()))
# 检查是否达到批量大小
if len(queues[func_id]) >= max_batch_size:
_process_batch(func, queues, locks, timers, func_id)
# 启动定时器
elif timers[func_id] is None:
timer = threading.Timer(
batch_interval,
_process_batch,
args=[func, queues, locks, timers, func_id]
)
timer.start()
timers[func_id] = timer
return wrapper
return decorator
def _process_batch(func, queues, locks, timers, func_id):
"""处理批量任务"""
with locks[func_id]:
if not queues[func_id]:
return
batch = queues[func_id][:]
queues[func_id] = []
timers[func_id] = None
# 执行批量处理
results = []
for args, kwargs, timestamp in batch:
result = func(*args, **kwargs)
results.append(result)
return results
# 使用示例
@batch_sync(batch_interval=5, max_batch_size=5)
def sync_data(item_id):
"""需要同步的数据处理函数"""
print(f"同步数据: {item_id}")
return f"processed_{item_id}"
# 模拟低频调用
for i in range(8):
sync_data(i)
time.sleep(1)
上下文管理器方式
from contextlib import contextmanager
import threading
from collections import namedtuple
import time
BatchContext = namedtuple('BatchContext', ['items', 'commit'])
class BatchSyncContextManager:
def __init__(self, batch_interval=5, max_batch_size=10):
self.batch_interval = batch_interval
self.max_batch_size = max_batch_size
self.buffer = []
self.lock = threading.Lock()
self.condition = threading.Condition(self.lock)
self._stop_event = threading.Event()
# 启动后台线程
self._worker = threading.Thread(target=self._worker_loop, daemon=True)
self._worker.start()
def __enter__(self):
"""进入上下文"""
return self
def __exit__(self, exc_type, exc_val, exc_tb):
"""退出上下文时处理剩余任务"""
self.flush()
def add(self, item):
"""添加单个项目到批量"""
with self.lock:
self.buffer.append(item)
if len(self.buffer) >= self.max_batch_size:
self.condition.notify() # 通知工作线程
def flush(self):
"""强制刷新缓冲区"""
with self.lock:
batch = list(self.buffer)
self.buffer.clear()
if batch:
self._process_batch(batch)
def _worker_loop(self):
"""后台工作线程"""
while not self._stop_event.is_set():
with self.lock:
# 等待通知或超时
self.condition.wait(self.batch_interval)
if not self.buffer:
continue
batch = list(self.buffer)
self.buffer.clear()
# 处理批次
self._process_batch(batch)
def _process_batch(self, batch):
"""处理批次数据"""
print(f"批量处理 {len(batch)} 个项目: {batch}")
# 实际同步逻辑...
def stop(self):
"""停止工作线程"""
self._stop_event.set()
self.condition.notify()
# 使用示例
with BatchSyncContextManager(batch_interval=5, max_batch_size=3) as manager:
for i in range(7):
manager.add(f"item_{i}")
time.sleep(1)
使用建议
-
选择合适的策略：
- 对实时性要求高：使用较小的
batch_interval - 任务量稳定：使用
max_batch_size控制 - 任务稀疏：增加
batch_interval
- 对实时性要求高：使用较小的
-
错误处理：
def _process_batch(self, batch): """带重试的批量处理""" max_retries = 3 for attempt in range(max_retries): try: # 执行同步 self._do_sync(batch) break except Exception as e: if attempt == max_retries - 1: # 记录失败的任务 self._log_failed_batch(batch, e) time.sleep(attempt * 2) # 指数退避 -
监控和日志：
import logging from datetime import datetime
def _process_batch(self, batch): start_time = datetime.now() try:
执行同步
result = self._sync(batch)
duration = (datetime.now() - start_time).total_seconds()
logging.info(f"Batch {len(batch)} items synced in {duration:.2f}s")
except Exception as e:
logging.error(f"Batch sync failed: {e}")
选择哪种方法取决于你的具体需求：
- **简单场景**：使用基础方案
- **高并发**：使用异步方案
- **代码整洁**：使用装饰器
- **资源管理**：使用上下文管理器
