一、为什么需要批处理#

你的应用需要往数据库插入 1000 条记录。如果逐条执行，每条 INSERT 都是一次网络往返、一次 SQL 解析、一次事务提交，1000 次开销叠加，总耗时可能达到几十秒。而一次批量 INSERT 只要一次网络往返、一次解析、一次提交，耗时可能不到一秒。

逐条处理的问题不只是慢。频繁的小操作还会给下游系统带来巨大压力：每次请求都有固定的上下文切换、连接获取、协议封装等开销。当吞吐量是瓶颈时，这些固定开销就会被无限放大。

把开销拆开来看。一次数据库写入的耗时大致分为两部分：固定开销和有效负载。固定开销包括 TCP 握手（约 1-3ms）、SQL 解析与执行计划生成（约 0.5-2ms）、事务提交与 WAL 写入（约 1-5ms）。有效负载才是真正写入数据的时间，通常不到 0.1ms。也就是说，一条 INSERT 的总耗时中，固定开销可能占 95% 以上。逐条执行 1000 次时，这 95% 的固定开销被重复了 1000 次；批量执行时，它只发生一次，被 1000 条记录分摊。

网络请求中这个比例更夸张。一次 HTTP 请求的固定开销包括 DNS 解析、TLS 握手、TCP 慢启动，加起来可能 50-200ms，而实际传输一个小的 JSON 体可能只要 1ms。调用 100 次外部 API，逐次调用意味着 5-20 秒花在连接建立上，批量接口可能一次搞定。

批处理的思路是把多个小操作攒成一个大操作一起执行，把每次操作的固定开销分摊到整个批次上。代价是单个操作要等一会儿才能被执行，增加了延迟；收益是整体吞吐量大幅提升。这是延迟和吞吐量之间的经典权衡：你用少量延迟换取大量吞吐。在写入密集型场景中，这个交换几乎总是划算的，一条记录晚 50ms 写入数据库，用户根本感知不到；但吞吐量从 100 QPS 提升到 5000 QPS，系统能承载的流量就完全不同了。

二、现实类比#

装洗碗机。你不会洗一个盘子就开一次机，整天的碗碟攒起来，一次洗完。每个盘子分摊的水费、电费、时间都远低于单独清洗。洗碗机不会因为你只有一个盘子就提前启动，它有自己的”批次大小”或”最长等待时间”。

这个类比还揭示了批处理的两个关键参数：洗碗机有容量上限（对应 maxSize），也有你的耐心上限，你不会让一个脏碗在池子里等三天（对应 maxWait）。容量满了就启动，时间到了也启动，两者缺一不可。

三、核心思想#

批处理器维护一个待处理队列，按照两种策略触发刷新：

1. 大小触发：队列中的项数达到 maxSize 时立即刷新
2. 时间触发：距离上次刷新超过 maxWait 时强制刷新，避免项目在队列中无限等待

关键属性：

触发策略对比：

四、变体与对比#

批处理和背压经常一起出现：批处理提高了消费速率，背压在消费速率仍不够时限制生产速率。当批次中某一条失败时，可以对该条单独进行指数退避重试，而不需要重试整个批次。

五、多语言实现#

1
package batch
2

3
import (
4
  "sync"
5
  "time"
6
)
7

8
// BatchProcessor 批处理器
9
type BatchProcessor[T any, R any] struct {
10
  queue    []batchEntry[T, R]
11
  process  func([]T) []R
12
  maxSize  int
13
  maxWait  time.Duration
14
  mu       sync.Mutex
15
  timer    *time.Timer
16
}
17

18
type batchEntry[T any, R any] struct {
19
  item T
20
  ch   chan R
21
}
22

23
func New[T any, R any](
24
  process func([]T) []R,
25
  maxSize int,
26
  maxWait time.Duration,
27
) *BatchProcessor[T, R] {
28
  return &BatchProcessor[T, R]{
29
    process: process,
30
    maxSize: maxSize,
31
    maxWait: maxWait,
32
  }
33
}
34

35
// Add 添加一项到待处理队列
36
func (bp *BatchProcessor[T, R]) Add(item T) R {
37
  bp.mu.Lock()
38
  defer bp.mu.Unlock()
39

40
  ch := make(chan R, 1)
41
  bp.queue = append(bp.queue, batchEntry[T, R]{item, ch})
42

43
  // 大小触发：队列满则立即刷新
44
  if len(bp.queue) >= bp.maxSize {
45
    bp.flush()
46
  } else if bp.timer == nil {
47
    // 时间触发：启动倒计时
48
    bp.timer = time.AfterFunc(bp.maxWait, func() {
49
      bp.mu.Lock()
50
      bp.timer = nil
51
      bp.flush()
52
      bp.mu.Unlock()
53
    })
54
  }
55

56
  return <-ch
57
}
58

59
// flush 刷新：取出队列中所有项，批量处理
60
func (bp *BatchProcessor[T, R]) flush() {
61
  if len(bp.queue) == 0 {
62
    return
63
  }
64
  items := make([]T, len(bp.queue))
65
  for i, e := range bp.queue {
66
    items[i] = e.item
67
  }
68
  results := bp.process(items)
69
  for i, e := range bp.queue {
70
    e.ch <- results[i]
71
  }
72
  bp.queue = bp.queue[:0]
73
}

1
class BatchProcessor<T, R> {
2
  private queue: Array<{ item: T; resolve: (r: R) => void }> = [];
3
  private timer: ReturnType<typeof setTimeout> | null = null;
4

5
  constructor(
6
    private processBatch: (items: T[]) => Promise<R[]>,
7
    private maxSize: number = 10,
8
    private maxWaitMs: number = 50,
9
  ) {}
10

11
  async add(item: T): Promise<R> {
12
    return new Promise<R>((resolve) => {
13
      this.queue.push({ item, resolve });
14

15
      // 大小触发：队列满则立即刷新
16
      if (this.queue.length >= this.maxSize) {
17
        this.flush();
18
      } else if (!this.timer) {
19
        // 时间触发：启动倒计时
20
        this.timer = setTimeout(() => this.flush(), this.maxWaitMs);
21
      }
22
    });
23
  }
24

25
  private async flush(): Promise<void> {
26
    if (this.timer) {
27
      clearTimeout(this.timer);
28
      this.timer = null;
29
    }
30

31
    // 取出当前队列的所有项
32
    const batch = this.queue.splice(0);
33
    if (batch.length === 0) return;
34

35
    // 批量处理
36
    const results = await this.processBatch(batch.map((b) => b.item));
37
    batch.forEach((b, i) => b.resolve(results[i]!));
38
  }
39
}

六、生产验证#

七、小结#

何时使用：

• 数据库写入，批量 INSERT 替代 N 次单条 INSERT，吞吐量提升数倍。固定开销从 N 份降为 1 份，最直接也最普遍的收益
• API 批量调用，减少网络往返，GraphQL DataLoader、Elasticsearch Bulk API 的核心。DataLoader 把 N 次单独查询合并为一次 WHERE id IN (...)，Elasticsearch 把 N 次索引请求合并为一次 Bulk 请求
• 消息队列，Kafka 按分区批处理、SQS 批量发送/接收。生产者攒批减少网络请求，Broker 顺序写入批量消息减少磁盘寻道
• UI 更新，React 批量 setState、浏览器批量 DOM 重排。5 次 setState 没有批处理就是 5 次渲染，有批处理就是 1 次

何时不用：

• 延迟敏感路径，批处理天然增加等待时间，实时系统不适用。交易下单、实时报价这类场景，50ms 的等待就不可接受
• 极低流量，很少超过 1 个项时，批处理增加复杂性却无收益。内部管理后台每天几十次操作，省下的毫秒不值得多维护一个队列和定时器
• 严格原子性要求，批次中某条失败时需要逐条错误处理，增加复杂度。业务要求”全成功或全失败”时，逐项错误处理不适用，原子性批处理又面临”一条坏数据拖垮整批”的风险
• 无界内存风险，没有大小限制时，流量高峰期队列可能无限增长。maxSize 不只是触发阈值，更是内存安全阀

八、参考资料#

• Kafka Producer Internals - Kafka 生产者批处理配置，含 batch.size 和 linger.ms
• DataLoader - GraphQL 批量加载库，批处理 + 缓存的经典实现
• Elasticsearch Bulk API - 批量索引文档的 API，含部分失败处理
• React Automatic Batching - React 18 自动批处理的官方说明
• Linux Block Layer Merging - Linux 块层请求合并机制文档