3. 合并批处理核心代码实现（Python/Node.js）

3.1 原理：队列缓冲区 + 定时器触发 + 批量发送

批处理请求合并的核心机制由三个组件构成：

  • 队列缓冲区：临时存放待发送的请求对象（含回调），避免逐个发起请求。
  • 定时器触发：固定时间窗口（如100ms），到期后清空缓冲区并触发批量发送。
  • 批量发送逻辑：将队列中的多个请求合并为一个网络请求体，发送后端再分发结果。

三个组件的协作决定了合并粒度与功耗平衡效果。下表对比了两种语言在实现上的核心差异：

组件Python (asyncio)Node.js (EventLoop)
队列结构asyncio.Queue普通 Array + 互斥锁
定时器asyncio.create_task + asyncio.sleep 循环setInterval / setTimeout
批量发送aiohttp.ClientSession.postaxios.post / fetch
回调分发Future / 事件循环回调Promise.resolve
关键设计原则：缓冲区的生命周期与定时器严格同步——定时器触发前必须确保“暂停”新的入队，避免竞争条件。

3.2 实现流程

  1. 初始化空队列与定时器状态变量。
  2. 暴露入队接口（enqueue(request)），将请求对象连同回调/ Promise 存入队列。
  3. 定时器每隔 batchWindow 毫秒触发一次：
    a. 从队列中一次性取出所有待处理请求。
    b. 组装成批量数据结构（如 JSON 数组）。
    c. 发起 HTTP POST 到批处理端点。
    d. 将服务端返回的每个结果分配到对应的回调/Promise 中。
  4. 处理异常：网络失败时重试或直接 reject 所有等待请求。
功耗平衡技巧：动态调整 batchWindow——当队列积压速率升高时缩短窗口，低负载时延长以节省无线模块唤醒次数。

3.3 Python 实现（asyncio）

import asyncio
import aiohttp

class BatchProcessor:
    def __init__(self, batch_url, window_ms=100):
        self.queue = asyncio.Queue()
        self.batch_url = batch_url
        self.window = window_ms / 1000.0
        self._running = True
        asyncio.create_task(self._timer_loop())

    async def enqueue(self, payload):
        """返回一个 Future，当批量处理完成后 resolve"""
        future = asyncio.get_event_loop().create_future()
        await self.queue.put((payload, future))
        return future

    async def _timer_loop(self):
        while self._running:
            await asyncio.sleep(self.window)
            batch = []
            while not self.queue.empty():
                item = self.queue.get_nowait()
                batch.append(item)
            if batch:
                asyncio.create_task(self._send_batch(batch))

    async def _send_batch(self, batch):
        payloads = [item[0] for item in batch]
        futures = [item[1] for item in batch]
        try:
            async with aiohttp.ClientSession() as session:
                async with session.post(self.batch_url, json=payloads) as resp:
                    results = await resp.json()
            for i, res in enumerate(results):
                if i < len(futures):
                    futures[i].set_result(res)
        except Exception as e:
            for f in futures:
                if not f.done():
                    f.set_exception(e)

    def stop(self):
        self._running = False

3.4 Node.js 实现（无第三方库依赖，核心逻辑）

class BatchProcessor {
  constructor(batchUrl, windowMs = 100) {
    this.batchUrl = batchUrl;
    this.queue = [];
    this.windowMs = windowMs;
    this._running = true;
    this._timer = setInterval(() => this._flush(), this.windowMs);
  }

  enqueue(payload) {
    return new Promise((resolve, reject) => {
      this.queue.push({ payload, resolve, reject });
    });
  }

  async _flush() {
    const batch = this.queue.splice(0, this.queue.length);
    if (batch.length === 0) return;
    try {
      const response = await fetch(this.batchUrl, {
        method: 'POST',
        headers: { 'Content-Type': 'application/json' },
        body: JSON.stringify(batch.map(item => item.payload))
      });
      const results = await response.json();
      batch.forEach((item, i) => {
        if (i < results.length) {
          item.resolve(results[i]);
        } else {
          item.reject(new Error('Result count mismatch'));
        }
      });
    } catch (err) {
      batch.forEach(item => item.reject(err));
    }
  }

  stop() {
    this._running = false;
    clearInterval(this._timer);
  }
}
常见错误：队列处理间隙的数据丢失
在 _flush 执行期间，新的入队操作可能仍在向同一数组 push。Python 的 asyncio.Queue 天然保护临界区；Node.js 需确保 splice 原子性，或使用 while (this.queue.length) + 锁（如 async-mutex）来避免。

3.5 调试与验证

  • 日志打印：在入队、定时触发、发送前后输出时间戳与队列长度。
  • 模拟服务端：本地启动一个简易 HTTP 端点，返回固定延迟与递增 ID，验证结果顺序。
  • 边界条件：测试高并发下队列积压超过 MAX_BATCH_SIZE 时的分片逻辑（通常需结合定时与数量双重阈值）。

3.6 性能与稳定性优化

优化点措施影响
缓冲区容量上限当队列长度 > N 时立即强制 flush降低背压延迟
批量发送超时控制为 HTTP 请求设置合理的 timeout防止单次失败阻塞全部
指数退避重试批量发送失败后，将失败的请求重新入队并加倍下次窗口网络抖动下提升吞吐
CPU 空闲检查仅在 Node.js 事件循环空闲时触发 flush减少非必要唤醒

3.7 核心流程图

批处理请求合并流程 批处理核心流程：队列 + 定时器 + 批量发送 外部请求入队 队列缓冲区 定时器(batchWindow) 批量HTTP请求 结果分发至回调 定时器到期后： 1. 取出所有待处理请求 2. 合并为 JSON 数组 3. 一次性 POST 到后端

3.8 部署注意事项

  • 在容器化环境（Docker）中需注意定时器精度受系统时钟跳动影响，应使用高精度计时器（Node.js 使用 setTimeout 链式调用而非 setInterval 避免漂移）。
  • 批量请求的 body 大小可能超出服务端限制（如 Nginx 默认 1MB），需配置 client_max_body_size 或在客户端分片。
  • 若批量端点无结果排序保证，需在请求 payload 中携带关联 ID，并在分发时根据 ID 匹配回调。
← 上一章 📖 返回目录 下一章 →