如何以池化方式消费Go语言无缓冲Channel？

需求说明 #

使用固定数量（n个）的Worker组成线程池消费消息队列，仅当有Worker空闲时才从队列获取新消息，避免一次性拉取全部消息。例如队列有100条消息时，先获取n条（如3条）处理，任意一条处理完成后再获取下一条，循环往复。

正确实现方案 #

核心思路是预先启动固定数量的Worker Goroutine，每个Worker持续从消息通道接收消息并处理。利用Go通道的阻塞特性，天然实现“空闲Worker自动获取新消息”的逻辑：

import "sync"

// 假设workerCount是你设定的Worker数量
const workerCount = 3

func consumeMessages(ctx context.Context, qName string) error {
    // 初始化消费通道，deliveries为<-chan Message类型
    deliveries, err := qb.pubSubSubscriber.Consume(ctx, qName)
    if err != nil {
        return err
    }

    // 启动worker池
    var wg sync.WaitGroup
    wg.Add(workerCount)

    for i := 0; i < workerCount; i++ {
        go func(workerID int) {
            defer wg.Done()
            // 每个worker持续从通道接收消息，无消息时阻塞
            for msg := range deliveries {
                // 处理消息
                result := myFunc(msg)
                // 可根据业务需求添加消息确认、结果处理逻辑
                // msg.Ack()
            }
        }(i)
    }

    // 等待所有worker退出（当ctx被取消或deliveries通道关闭时）
    wg.Wait()
    return nil
}

方案说明 #

1. Worker启动逻辑：预先启动workerCount个Goroutine，每个Worker进入循环监听deliveries通道。
2. 自动控流机制：当deliveries有消息时，空闲的Worker会优先接收消息并处理；所有Worker都忙碌时，新消息会在通道中等待（通道带缓冲）或阻塞消息生产者（通道无缓冲），完全匹配“仅空闲时取新消息”的需求。
3. 优雅退出：通过sync.WaitGroup等待所有Worker完成，当ctx被取消或队列消费结束（deliveries通道关闭）时，Worker循环自动退出，所有Goroutine优雅结束。

为什么之前的方案无效？ #

• Tunny池方案：先通过for msg := range deliveries一次性拉取所有消息到本地内存，再分配给Worker池处理，本质是先把消息从队列全量拉取，再做本地分发，不符合“仅空闲时取新消息”的要求。
• 自制Counter+WaitGroup方案：存在并发安全问题（counter未用原子操作保护），且WaitGroup逻辑错误（wg.Wait()会阻塞直到所有Done()调用，无法实现“单个Worker空闲就取新消息”的效果），逻辑混乱易引发死锁或消息丢失。

内容的提问来源于stack exchange，提问作者Lukas Müller