如果你想只在进程中按标签执行FIFO队列，你可以为每个标签分配一个通道和一个工作器。你可以在Map中跟踪每个标签的通道。
这里有一个实现（https://go.dev/play/p/8r8-oaRaY2x），它非常适合你，你的实际通道消息可以包括一个结果通道。

package main

import (
    "sync"
    "log"
)

type Route[T any] struct {
    Input  chan []T
    Output chan T
    Worker func(<-chan T, *sync.WaitGroup)
}

func (r *Route[T]) Accept(wg *sync.WaitGroup) {
    defer wg.Done()
    for inputs := range r.Input {
        for _, input := range inputs {
            r.Output <- input
        }
    }
    close(r.Output)
}

func (cr *Route[T]) enqueue(messages []T, wg *sync.WaitGroup) {
    defer wg.Done()
    cr.Input <- messages
}

func (r *Route[T]) Start(wg *sync.WaitGroup) *Route[T] {
    wg.Add(2)
    go r.Accept(wg)
    go r.Worker(r.Output, wg)
    return r
}

func (r *Route[T]) Close() {
    close(r.Input)
}

type Router[T any] struct {
    sync.Mutex
    routes map[string]*Route[T]
    wg     sync.WaitGroup
    routers_wg sync.WaitGroup
    routemaker func(tag string)*Route[T]
}

func (cr *Router[_]) Close() {
    for _, r := range cr.routes {
        r.Close()
    }
}

func (cr *Router[_]) Wait() {
    cr.routers_wg.Wait()
    cr.wg.Wait()
}

func (cr *Router[T]) Route(tag string, messages []T, wg *sync.WaitGroup) {
    cr.Lock()
    defer cr.Unlock()
    for m, r := range cr.routes {
        if m == tag {
            go r.enqueue(messages, wg)
            return
        }
    }
    new_route := cr.routemaker(tag).Start(&cr.routers_wg)
    cr.routes[tag] = new_route
    go new_route.enqueue(messages, wg)
}

type worker[T any] interface {
    Work(<-chan T)
}

type collateWork[T any] struct {
    work []T
    tag string
}

func (cw *collateWork[T]) Work(c <-chan T, wg *sync.WaitGroup) {
    defer wg.Done()
    for w := range c {
        cw.work = append(cw.work, w)
    }
    log.Printf("Worker '%s' collation: %v", cw.tag, cw.work)
}

func main() {

    var routes = Router[int]{
        routes: make(map[string]*Route[int]),
        routemaker: func(tag string) *Route[int] {
            return &Route[int]{
                Input:  make(chan []int, 30),
                Output: make(chan int, 30),
                Worker: (&collateWork[int]{tag: tag}).Work,
            }
        },
    }
    var router_waits sync.WaitGroup
    router_waits.Add(6)
    go routes.Route("a", []int{1, 2, 3, 4, 5, 6, 7}, &router_waits)
    go routes.Route("b", []int{10, 20, 30, 40, 50, 60, 70}, &router_waits)
    go routes.Route("a", []int{8, 9, 10, 11, 12, 13, 14, 15}, &router_waits)
    go routes.Route("b", []int{100, 200, 300, 400, 500, 600, 700}, &router_waits)
    go routes.Route("c", []int{1000, 2000, 3000, 4000, 5000, 6000, 7000}, &router_waits)
    go routes.Route("c", []int{8000, 9000 }, &router_waits)
    router_waits.Wait()
    routes.Close()
    routes.Wait()
}

个字符