Go 实现限并发Worker池 支持单任务超时自动跳过慢任务

现有实现的核心问题拆解 #

• 第一版代码的结果收集逻辑完全错位：超时控制的粒度本该是单个任务的计算耗时，代码却把超时加在了主协程等待结果的流程上。每轮收集结果都单独起一个1秒超时的定时器，只要任意一轮没立刻拿到结果就退出循环，哪怕jobs通道里还有大量未处理、未返回的任务，自然会丢失慢任务对应的结果。
• 加context后的版本会挂死的原因有两个：
  1. context完全没有被计算逻辑感知：模拟慢任务的time.Sleep(5*time.Second)是不可中断的，哪怕context的超时时间已经到了，Sleep也会硬等5秒才会走到后面的select分支，根本不会触发超时跳过的逻辑。
  2. defer c()的作用域完全错误：defer是函数级别延迟执行的，写在for循环里的defer，会把每一轮的cancel调用都压入栈，直到整个worker函数退出才会执行，既会造成context资源泄漏，也会因为主协程提前退出、没人读结果通道，导致worker发结果的时候永久阻塞。

实现的核心原则 #

Go语言没有提供强制终止任意运行中goroutine的原生能力，所谓“终止当前计算流程、跳去处理下一个任务”，本质是通过可中断的信号传递，让正在执行的计算逻辑主动感知超时状态、立刻停止后续计算。
要实现不阻塞主worker循环的单任务超时，必须把耗时的计算逻辑和worker的循环解耦，不能让计算逻辑的阻塞直接拖住worker读jobs通道的流程。

正确实现的关键步骤 #

• 给每个任务单独创建带超时时间的context，作为当前任务的超时信号源
• 把具体的计算逻辑放到独立的子goroutine中运行，计算逻辑内部必须主动监听ctx.Done()的信号，一旦超时立刻退出计算流程，不要继续执行无意义的计算
• worker主循环只监听两个事件：计算完成返回结果、任务超时触发。不管是正常完成还是超时，都立刻跳过当前任务，进入下一轮从jobs通道读任务的流程，不会被单个慢任务拖住
• 每轮任务处理完（无论成功/超时）立刻手动调用context的cancel方法释放资源，不要用循环内的defer，避免资源堆积
• 所有worker任务处理完成后，主动关闭结果通道，主协程只需要持续读取结果通道直到通道关闭，就可以拿到全部有效结果，不需要额外加等待超时

核心代码参考 #

func doStuff(){
    rules := []string{
        "a", "b", "c", "d", "e", "f", "g",
    }
    var (
        jobs    = make(chan []string, len(rules))
        res     = make(chan string, len(rules))
        matches []string
    )

    // 修正后的worker
    w := func(jobs <-chan []string, results chan<- string) {
        for j := range jobs {
            // 单任务1秒超时
            ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
            k, id := j[0], j[1]

            taskDone := make(chan string, 1)
            // 独立goroutine跑具体计算，避免拖住worker主循环
            go func() {
                if id == "c" || id == "e" {
                    // 把原来的不可中断Sleep换成可响应超时的等待
                    select {
                    case <-time.After(5 * time.Second):
                        // 正常等待结束，走后续计算
                    case <-ctx.Done():
                        // 超时触发，立刻退出计算
                        return
                    }
                }
                m := match(k, id)
                taskDone <- m
            }()

            // worker主循环等待结果或超时
            select {
            case m := <-taskDone:
                results <- m
            case <-ctx.Done():
                fmt.Println("task timeout, skip:", id)
            }
            // 立刻释放context资源
            cancel()
        }
        // 所有任务处理完，关闭结果通道
        close(results)
    }

    // 启动N个固定数量的worker
    N := 2
    for i := 0; i < N; i++ {
        go w(jobs, res)
    }

    // 投递所有任务
    for _, rl := range rules {
        jobs <- []string{"a", rl}
    }
    close(jobs)

    // 主协程持续收集结果，直到通道关闭
    for match := range res {
        matches = append(matches, match)
    }
    fmt.Println(matches) // 最终输出 [a b d f g]
}

常见的坑点提醒 #

• 不要把不可中断的耗时逻辑（比如无超时的SQL查询、硬Sleep、无监听的死循环计算）直接写在worker主循环里，这类逻辑永远会被超时信号卡住，根本达不到跳过的效果。所有耗时的外部调用，都必须支持传入context、能被超时中断。
• 不要把超时判断的select写在完整计算流程的后面，相当于先硬跑完所有计算，才去检查有没有超时，完全失去了超时控制的意义——等你把5秒的慢计算跑完，再触发跳过，计算的耗时已经花出去了，根本达不到快速跳过的效果。

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