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
}