Async Cargo 自适应批处理

示例说明

埋点上报、日志聚合这类场景希望攒批降低调用次数，但事件到达速率不均。cargo 的队列在达到阈值时立即冲刷一批，未达标时后续事件继续排队，处理器同一时间只处理一个批次，批次间天然串行。push 是非阻塞的，生产者永远不需要等待消费者。

技术要点

• cargo 第二参数是批大小阈值
• push 支持单条或数组批量入队
• drain 等待队列彻底清空

代码解读

flush 模拟批量写入耗时 30 毫秒；前三个事件入队后立即凑满一批触发冲刷，第四个开启新批次。数组形式的 push 一次加入两条并附带完成回调。drain 返回的 Promise 在所有批次处理完 resolve，适合在关机前等待缓冲排空，idle 判断队列是否空闲。

使用建议

批大小要参考下游批量接口的吞吐最优区间。需要并发处理多个批次时改用 cargo 的并发版本。优雅停机流程应先停止入队再 drain，避免丢尾部数据。

代码示例

import cargo from 'async/cargo';

const flush = async (items) => {
  console.log('flushing', items.length, 'events');
  await new Promise((r) => setTimeout(r, 30));
};

const queue = cargo(flush, 3);

queue.push('event-1');
queue.push('event-2');
queue.push('event-3');
queue.push('event-4');

queue.push(['event-5', 'event-6'], (err) => {
  console.log('callback after group drained, err:', err ?? null);
});

const drained = await queue.drain();
console.log('all batches processed:', drained === undefined);

queue.push('event-7');
await queue.idle();