Python脚本如何合并低频多次同步任务

wen python案例 2026-06-25 254

定时批量合并（基础方案）

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}")


选择哪种方法取决于你的具体需求：
- **简单场景**：使用基础方案
- **高并发**：使用异步方案
- **代码整洁**：使用装饰器
- **资源管理**：使用上下文管理器

本文地址： https://www.bmjup.com/post/8194.html