如何处理共享同一channel的多个goroutine的消息分发与重入问题

问题1 阻塞原因 #

• 核心死锁逻辑：你创建的requeue是无缓冲channel，只有读端就绪时写入才会成功。同时公共消费channelch仅设置了2个缓冲位，高并发下大量不匹配的数据会被处理协程写入requeue，而负责将requeue数据回写ch的goroutine会因为ch已满被阻塞，此时requeue的读端相当于停止工作，所有处理协程往requeue写数据的操作会全部卡住，直接形成死锁。
• 隐藏异常点：你在wg.Wait()后直接关闭了ch，但此时requeue中可能还有未处理的数据，重入goroutine尝试往已关闭的ch写数据会触发panic，也会导致流程中断。

问题2 更优的处理方案 #

你当前的多数据集混发+重入匹配的模式本身就存在无效开销和死锁风险，更合理的方案是新增数据路由层，完全移除requeue逻辑：

1. 为每个数据集对应的处理协程创建独立的专属channel，不同数据集的数据互不干扰
2. 启动一个独立的路由goroutine，接收所有API返回的原始数据，直接根据数据的dataset字段转发到对应处理协程的专属channel
3. 每个处理协程只需要从自己的专属channel读取数据，无需做归属校验，也没有重入需求，既减少了无效计算，也彻底避免了重入队列的阻塞问题

核心逻辑改造示例 #

// 先创建dataset到专属channel的映射
datasetChMap := make(map[string]chan dweather.DWeatherResponse)
final := make(chan strike.StrikePerYearResponse)
// 初始化每个处理协程的专属channel
for _, s := range args.Parameters.Strikes {
    dataset := string(s.Dataset)
    ch := make(chan dweather.DWeatherResponse, 100) // 可根据数据量调整缓冲大小
    datasetChMap[dataset] = ch
    strike := strike.StrikePerYear{
        Parameter: strike.Parameter(s.Dataset),
        StrikeValue: s.Value,
    }
    // 处理协程只从自己的专属channel读数据，不需要校验归属和重入
    go strike.ProcessStrikePerYear(ch, final, dataset)
}

// 公共路由channel，接收所有API返回的数据
routerCh := make(chan dweather.DWeatherResponse, 100)
// 启动路由goroutine做数据分发
go func() {
    for data := range routerCh {
        if ch, ok := datasetChMap[data.Dataset]; ok {
            ch <- data
        }
    }
    // 所有数据分发完成后关闭所有专属channel
    for _, ch := range datasetChMap {
        close(ch)
    }
}()

// API请求的goroutine把数据都发到路由channel
var wg sync.WaitGroup
for _, dataset := range api.Params.Dataset {
    wg.Add(1)
    go api.RequestData(ctx, routerCh, dataset, &wg)
}
wg.Wait()
close(routerCh) // 通知路由goroutine数据已全部发送完成

// 后续汇总结果逻辑不变
var strikes []strike.StrikePerYearResponse
for range args.Fetch.Datasets {
    strikes = append(strikes, <-final)
}
return strikes

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