一个「巧思」线程池：用单协程无锁控 Worker，动态扩缩容 + 优雅关闭全拿捏

Hyman 2025/11/17 实践总结最佳

# 前言

在并发编程的世界里，线程池（或协程池）是控制资源开销、提升任务执行效率的核心组件。但传统线程池设计中，Worker 数量的动态调整往往伴随着复杂的锁竞争，扩缩容逻辑绕不开互斥锁的保护；结果回传要么依赖阻塞等待，要么需要复杂的回调绑定；优雅关闭更是容易出现任务丢失或资源泄露的坑。

今天要聊的这个协程池实现，却用一个「巧思」解决了这些问题：用单协程作为控制中枢，无锁管理 Worker 生命周期，同时天然支持动态扩缩容、异步结果回传和丝滑的优雅关闭。话不多说，我们来拆解它的设计智慧。（开源地址：https://github.com/itart-top/task-pool）

# 核心设计：单协程控制中枢，从根源避免锁竞争

传统线程池的 Worker 数量调整，通常需要多个线程同时操作 Worker 计数器，因此必须用互斥锁（如sync.Mutex）保护共享状态，这不仅增加了代码复杂度，还可能在高并发调整时引发性能损耗。

而这个实现的破局点在于：引入一个单独的controlLoop协程作为「控制中枢」，所有 Worker 的创建、销毁都由它独家负责。

func (p *Pool) controlLoop() {
    // 启动初始 workers
    p.startWorkers(p.workerCount)
    for {
        select {
        case n := <-p.resizeReq:  // 处理扩缩容请求
            p.doResize(n)
        case <-p.closeReq:        // 处理关闭请求
            // 取消所有Worker，触发优雅退出
            for _, cancel := range p.cancelFuncs {
                cancel()
            }
            return
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

controlLoop是整个池的「大脑」，它维护着workerCount（当前 Worker 数量）和cancelFuncs（每个 Worker 的取消函数）这两个核心状态。由于这两个状态只被controlLoop访问，完全不需要加锁 —— 从根源上避免了锁竞争。

这种设计的巧妙之处在于：所有对 Worker 数量的修改（扩缩容、关闭）都通过 channel（resizeReq、closeReq）通知controlLoop，由它串行处理。串行化的操作天然线程安全，代码也更简洁。

# 动态扩缩容：轻量高效，只响应最新需求

动态扩缩容是线程池应对流量波动的关键能力。这个实现的扩缩容逻辑非常轻量，且能「自动忽略过时请求」。

当调用Resize(n)时，请求会被发送到resizeReq通道，而通道的缓冲大小为 1：

func (p *Pool) Resize(n int) {
    if n <= 0 {
        n = 1
    }
    if p.closed.Load() {
        return
    }
    // 先清空旧请求，只保留最新的
    select {
    case <-p.resizeReq:
    default:
    }
    select {
    case p.resizeReq <- n:
    default:
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

这意味着：如果短时间内连续调用Resize（比如连续调整到 10、20、30），controlLoop只会处理最后一次请求（30），避免了频繁创建 / 销毁 Worker 的资源浪费。

而实际的扩缩容操作doResize则非常直接：

func (p *Pool) doResize(target int) {
    diff := target - p.workerCount
    p.workerCount = target
    if diff > 0 {
        p.startWorkers(diff)  // 新增Worker
    } else if diff < 0 {
        p.stopWorkers(-diff)  // 销毁多余Worker
    }
}

1
2
3
4
5
6
7
8
9

新增 Worker 时，startWorkers会创建新的协程，并通过context.WithCancel绑定取消函数，便于后续销毁；
销毁 Worker 时，stopWorkers会调用对应 Worker 的取消函数，触发 Worker 退出。

整个过程无需锁保护，因为workerCount和cancelFuncs的修改完全由controlLoop串行处理，安全且高效。

# 异步结果回传：订阅模式，灵活解耦

任务执行结果的回传是线程池的常见需求。传统做法要么让用户阻塞等待（如sync.WaitGroup+channel），要么在提交任务时绑定回调函数，灵活性较差。

这个实现采用了「订阅者模式」，通过Subscribe方法注册结果处理器，任务执行完毕后自动回调：

// 注册结果订阅者（并发安全）
func (p *Pool) Subscribe(sub ResultSubscriber) {
    p.subscriber.Store(sub)  // atomic.Value保证并发安全
}

// 任务执行后回调订阅者
func (p *Pool) invoke(task Entry) {
    // ... 执行任务 ...
    // 任务结束后，通过订阅者回传结果
    if v := p.subscriber.Load(); v != nil {
        if sub, ok := v.(ResultSubscriber); ok && sub != nil {
            sub(TaskResult{
                TaskID:   task.TaskID,
                Data:     data,
                Err:      err,
                Duration: time.Since(start),
            })
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

这种设计的优势在于：

解耦任务提交与结果处理：提交任务的代码无需关心结果如何处理，订阅者可以随时注册 / 修改；
并发安全：用atomic.Value存储订阅者，避免了回调函数注册时的锁竞争；
信息完整：结果中包含任务 ID、数据、错误和执行时间，便于追踪任务状态。

如果需要多订阅者，还可以基于这个基础扩展为发布 - 订阅模式（如维护一个订阅者列表），灵活性极高。

# 优雅关闭：不丢任务，不留资源

线程池的「优雅关闭」是个容易被忽略的细节 —— 如何保证关闭时正在执行的任务不中断、队列中剩余的任务不丢失，同时不泄露资源？

这个实现的Close方法堪称典范：

func (p *Pool) Close() {
    if !p.closed.CompareAndSwap(false, true) {
        return  // 避免重复关闭
    }
    close(p.closeReq)  // 通知controlLoop开始关闭
    p.wg.Wait()        // 等待所有Worker退出
}

1
2
3
4
5
6
7

关闭流程的核心逻辑在 Worker 的退出处理中：

func (p *Pool) worker(ctx context.Context) {
    defer p.wg.Done()
    for {
        select {
        case <-ctx.Done():
            // 如果是池关闭（closed==true），则尝试消费剩余任务
            if p.closed.Load() {
                goto flushTask
            }
            return  // 否则（如Resize缩容）直接退出
        case task, ok := <-p.tasks:
            if !ok {
                return
            }
            p.invoke(task)
        }
    }

flushTask:  // 关闭时的任务冲洗阶段
    for {
        select {
        case task, ok := <-p.tasks:
            if !ok {
                return
            }
            p.invoke(task)  // 处理剩余任务
        default:
            return  // 队列空了，退出
        }
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

当Close被调用后：

controlLoop会收到closeReq通知，调用所有 Worker 的取消函数，触发 Worker 退出；
Worker 检测到closed为true时，会进入flushTask阶段，逐个消费队列中剩余的任务，直到队列为空；
所有 Worker 执行完剩余任务并退出后，p.wg.Wait()才会返回，保证关闭操作完成后没有遗留任务和资源。

这种设计既避免了关闭时的任务丢失，又不会让 Worker 无限等待新任务，真正做到了「优雅」。

# 其他细节：稳健性拉满

除了核心设计，这个实现还有很多细节值得称道：

任务提交安全：Submit方法会检查池是否已关闭，或等待上下文取消，避免向已关闭的池提交任务；
Panic 防护：任务执行时通过defer recover()捕获恐慌，并将错误信息（含堆栈）通过结果回传，避免单个任务崩溃导致整个 Worker 退出；
默认值合理：创建池时如果传入无效的 Worker 数量或任务缓冲大小，会自动修正为合理默认值（1 个 Worker、100 个缓冲），降低使用门槛。

# 总结：巧思带来的简洁与高效

这个协程池的设计，用一个controlLoop协程作为控制中枢，巧妙地避免了锁竞争；通过 channel 传递控制信号，实现了轻量高效的动态扩缩容；用订阅者模式解耦结果处理，兼顾灵活性与并发安全；最后通过精细的关闭流程，保证了任务不丢失、资源不泄露。

它的代码量不大，但每一处设计都透着巧思 —— 用最简单的方式解决了传统线程池的复杂问题。这种「大道至简」的设计思路，值得我们在并发组件开发中借鉴。

如果你需要一个支持动态扩缩容、异步结果回传且能优雅关闭的协程池，这个实现绝对是个不错的选择。

艺术码农的小栈

Choose mode