Go协程批生成器未推送剩余批次至输出通道问题排查

核心原因 #

for input := range batcher.Input循环会持续阻塞接收数据，只有当输入通道被主动关闭后，循环才会退出。如果输入通道始终处于开放状态，代码永远执行不到循环后的剩余批次处理块，这就是[11,12]无法进入输出通道的根本原因。添加sync.WaitGroup无效，是因为WaitGroup只是等待goroutine执行完成，但如果循环一直阻塞，goroutine根本不会走到结束逻辑。

解决方案 #

1. 必须关闭输入通道 #

在所有需要发送到batcher.Input的数据发送完成后，调用close(batcher.Input)。这是让range循环退出、执行剩余批次处理的关键。

示例发送逻辑：

// 向输入通道发送所有数据
for _, num := range []int{1,2,3,4,5,6,7,8,9,10,11,12} {
    batcher.Input <- num
}
// 发送完成后关闭输入通道
close(batcher.Input)

2. 确保剩余批次处理逻辑正确 #

批处理goroutine中的剩余批次代码本身没问题，但要保证循环退出后能执行到：

func batchProcessor(batcher *Batcher) {
    batch := make([]int, 0, batcher.BatchSize)
    for input := range batcher.Input {
        batch = append(batch, input)
        // 达到批次大小则输出
        if len(batch) == batcher.BatchSize {
            batcher.Output <- batch
            batch = make([]int, 0, batcher.BatchSize)
        }
    }
    // 循环退出后处理剩余批次
    if len(batch) > 0 {
        batcher.Output <- batch
    }
}

3. 修正sync.WaitGroup的使用 #

如果用WaitGroup等待批处理完成，要确保WaitGroup覆盖完整的批处理流程（包括剩余批次的发送）：

var wg sync.WaitGroup
wg.Add(1)
go func() {
    defer wg.Done()
    batchProcessor(batcher)
}()

// 发送数据并关闭输入通道...

wg.Wait()
// 所有批处理完成后关闭输出通道（如果需要）
close(batcher.Output)

验证结果 #

调整后，输出会符合预期：

Output:[1 2 3 4 5]
Output:[6 7 8 9 10]
Output:[11 12]

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