基于批量处理的高效队列：Golang无延迟实现方案问询

我要在服务里实现一个批量处理的内部队列，客户端单次传入1个元素。核心需求：

• 多元素并行提交时批量处理，提升扩展性
• 不能为等待潜在新请求添加人为延迟，避免拖慢顺序请求的处理速度（延迟值难以确定，调优成本高，还要兼顾两种场景的性能）

Python里我是这么实现的：

BATCH_SIZE = 100

def drain_queue():
  ev = ev_queue.get()
  evs = []
  evs.append(ev)
  while True:
    if len(evs) == BATCH_SIZE:
      break
    try:
      ev = ev_queue.get(block=False)
    except Empty:
      break
    evs.append(ev)
  process_evs(evs)

现在要找Golang下基于channel的类似实现。我自己写的方案会在顺序请求场景下产生1秒延迟，不符合需求：

dur := time.Duration(1000) * time.Millisecond // 1 second wait.
for msg := range evQueue {
  // Start a new batch with this message.
  batch := []BatchArg{msg}
  timedOut := false
  // Loop until the batch is full or a time out.
  for timeout := time.After(dur); len(batch) < 200 && !timedOut; {
    select {
    case other := <-evQueue:
      batch = append(batch, other)
    case <-timeout:
      // The batching channel is empty, and the timer ticked.
      timedOut = true
    }
  }
  processBatch(batch)
}

需要无额外延迟的Golang实现方案。

可以用select配合default分支实现非阻塞读取channel，完全对齐Python版本的逻辑——先拿到第一个元素，再尽可能非阻塞读取剩余元素，直到批次满或channel暂时为空，既保证批量处理并行请求，又不会给顺序请求带来额外延迟。

代码实现如下：

const batchSize = 100

func drainQueue(evQueue <-chan BatchArg) {
    for {
        // 先阻塞读取第一个元素，避免空轮询
        ev, ok := <-evQueue
        if !ok {
            // channel被关闭，退出循环
            return
        }
        batch := []BatchArg{ev}
        
        // 非阻塞读取剩余元素，直到批次满或无元素
        for len(batch) < batchSize {
            select {
            case nextEv, ok := <-evQueue:
                if !ok {
                    // channel关闭，处理当前批次后退出
                    processBatch(batch)
                    return
                }
                batch = append(batch, nextEv)
            default:
                // 当前channel无元素，停止读取
                break
            }
        }
        
        // 处理当前批次
        processBatch(batch)
    }
}

逻辑说明 #

1. 外层循环先阻塞读取第一个元素，确保只有当有请求进来时才启动批次处理，不会出现无效空轮询
2. 内层循环通过select+default实现非阻塞读取：
   • 只要channel内还有待处理元素，就继续加入批次
   • 一旦channel暂时为空，立刻退出内层循环并处理当前批次，无任何等待延迟
3. 当channel被关闭时，会处理完剩余元素后再退出，避免数据丢失
4. 批次达到设定大小后立刻处理，保证批量处理的效率

该实现完全匹配你Python版本的核心逻辑，同时满足无延迟处理顺序请求、批量处理并行请求的需求。

内容的提问来源于stack exchange，提问作者likecs