如何在并行处理中实现同标签任务的FIFO串行执行？

核心思路是给每个标签分配专属任务通道，同标签任务全部发送到该通道，由单独goroutine按顺序处理；不同标签任务则由各自goroutine并行处理。同时用sync.Map动态管理这些标签对应的通道和goroutine，避免内存泄漏，完美适配99%标签唯一的场景。

修改后的代码如下：

import (
    "sync"
)

type task struct {
    tag  string
    data []int
}

// 全局管理每个标签对应的任务通道
var tagWorkers = sync.Map{}

func process(ch chan task) {
    for t := range ch {
        // 获取当前标签的任务通道
        workerCh, ok := tagWorkers.Load(t.tag)
        if !ok {
            // 标签不存在则创建新通道
            newCh := make(chan task, 10) // 缓冲区大小可根据业务调整
            workerCh, _ = tagWorkers.LoadOrStore(t.tag, newCh)
            // 启动该标签的专属处理goroutine
            go func(tag string, ch chan task) {
                defer func() {
                    // 任务处理完后删除标签记录，释放资源
                    tagWorkers.Delete(tag)
                }()
                // 按FIFO顺序处理所有任务
                for task := range ch {
                    processTask(task)
                }
            }(t.tag, newCh.(chan task))
        }
        // 将任务发送到对应标签的通道
        workerCh.(chan task) <- t
    }
}

func processTask(t task) {
    // 这里写你的任务处理逻辑
}

关键细节说明 #

• 同标签串行保障：每个标签对应唯一通道和处理goroutine，通道的FIFO特性天然保证任务顺序，且同一时间只有一个任务在处理。
• 动态资源回收：当某个标签的最后一个任务处理完成后，goroutine自动退出并删除标签记录，不会产生内存泄漏，适配大部分标签唯一的场景。
• 跨标签并行不阻塞：不同标签的任务由独立goroutine处理，完全并行，互相不影响。
• 缓冲区优化：任务通道设置缓冲区，可避免短时间内同标签任务集中到来时的发送阻塞，缓冲区大小可根据实际业务压力调整。

补充提示 #

如果需要优雅关闭所有处理goroutine，可以在process函数退出前，遍历sync.Map中的所有通道并关闭，确保goroutine正常退出。

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