线程池进阶：动态调整线程数、任务优先级队列、线程池的优雅关闭

线程池这东西，基础版本写起来不难。但真正用到生产环境，你会发现坑一个接一个。我早期做过一个后台服务，线程池大小写死了，结果高峰期请求量暴增，任务队列塞满，内存直接爆了。嗯，从那以后我就开始琢磨——线程池必须能动态调整。

这一章，咱们就聊聊线程池的三个进阶话题：动态调整线程数、任务优先级队列、优雅关闭。说白了，就是让线程池从「能用」变成「好用」。

一、动态调整线程数

为什么需要动态调整？你想想看，系统负载是波动的。白天高峰期线程不够用，任务排队；晚上低峰期线程太多，白白浪费资源。我习惯的做法是：让线程池根据任务队列长度或CPU使用率，自动增减线程数。

1.1 核心思路

动态调整的核心逻辑其实不复杂：

• 定期检查任务队列长度
• 如果队列积压超过阈值，增加工作线程
• 如果线程空闲时间过长，减少工作线程
• 设置上下限，防止线程数失控

1.2 代码实现

class DynamicThreadPool {
public:
    DynamicThreadPool(size_t min_threads, size_t max_threads)
        : min_threads_(min_threads)
        , max_threads_(max_threads)
        , current_threads_(min_threads)
        , stop_(false) {
        
        // 启动调整线程
        adjuster_ = std::thread([this] { adjust_loop(); });
        
        // 创建工作线程
        for (size_t i = 0; i < min_threads; ++i) {
            workers_.emplace_back([this] { worker_loop(); });
        }
    }

private:
    void adjust_loop() {
        while (!stop_) {
            std::this_thread::sleep_for(std::chrono::seconds(5));
            
            size_t queue_size = tasks_.size();
            size_t active = active_threads_.load();
            
            // 队列积压，增加线程
            if (queue_size > threshold_ && current_threads_ < max_threads_) {
                size_t to_add = std::min(queue_size / 10, max_threads_ - current_threads_);
                for (size_t i = 0; i < to_add; ++i) {
                    workers_.emplace_back([this] { worker_loop(); });
                }
                current_threads_ += to_add;
            }
            
            // 线程空闲，减少线程
            if (queue_size == 0 && active < current_threads_ / 2 
                && current_threads_ > min_threads_) {
                // 发送退出信号给空闲线程
                idle_exit_.store(true);
                idle_cv_.notify_all();
            }
        }
    }

    std::vector<std::thread> workers_;
    std::atomic<size_t> active_threads_{0};
    std::atomic<size_t> current_threads_;
    size_t min_threads_;
    size_t max_threads_;
    std::atomic<bool> stop_;
    std::atomic<bool> idle_exit_{false};
    std::condition_variable idle_cv_;
    static constexpr size_t threshold_ = 100;
};

二、任务优先级队列

有些场景下，任务有轻重缓急。比如一个日志系统，写错误日志的优先级应该高于写调试日志。普通的FIFO队列就不够用了。

2.1 优先级队列的设计

我建议用std::priority_queue作为底层容器，配合互斥锁实现线程安全。但要注意一点：优先级队列的top()返回的是最高优先级元素，而pop()移除它。这正好符合我们的需求。

template<typename T>
class PriorityTaskQueue {
public:
    void push(T task, int priority) {
        std::lock_guard<std::mutex> lock(mutex_);
        queue_.emplace(priority, std::move(task));
        cv_.notify_one();
    }

    T pop() {
        std::unique_lock<std::mutex> lock(mutex_);
        cv_.wait(lock, [this] { return !queue_.empty() || stop_; });
        
        if (stop_ && queue_.empty()) {
            return T{};  // 哨兵值
        }
        
        T task = std::move(queue_.top().second);
        queue_.pop();
        return task;
    }

private:
    std::priority_queue<std::pair<int, T>> queue_;
    std::mutex mutex_;
    std::condition_variable cv_;
    bool stop_ = false;
};

2.2 优先级反转问题

这里有个坑——优先级反转。什么意思？就是高优先级任务被低优先级任务阻塞了。比如线程池所有线程都在处理低优先级任务，高优先级任务只能排队等着。

解决办法有两个：

• 预留几个「高优先级专用线程」，只处理高优先级任务
• 任务抢占：当高优先级任务入队时，中断当前低优先级任务

我个人更倾向于第一种方案，简单可靠。第二种方案实现复杂，而且中断任务可能导致数据不一致。

三、线程池的优雅关闭

线程池关闭这件事，看似简单，实则暗藏杀机。直接std::terminate？那正在执行的任务怎么办？队列里还没执行的任务怎么办？

3.1 关闭的三个阶段

我总结了一个三阶段关闭法：

1. 拒绝新任务：设置标志位，push()直接返回false
2. 等待执行中的任务完成：让工作线程把当前任务跑完
3. 清理未执行的任务：队列里剩下的任务，要么丢弃，要么持久化

void shutdown() {
    // 第一阶段：拒绝新任务
    stop_accept_.store(true);
    
    // 第二阶段：通知所有工作线程准备退出
    stop_.store(true);
    cv_.notify_all();
    
    // 第三阶段：等待所有线程结束
    for (auto& worker : workers_) {
        if (worker.joinable()) {
            worker.join();
        }
    }
    
    // 第四阶段：处理剩余任务
    std::lock_guard<std::mutex> lock(mutex_);
    while (!tasks_.empty()) {
        auto task = std::move(tasks_.front());
        tasks_.pop();
        // 这里可以记录日志，或者持久化到磁盘
        // 我习惯把未执行的任务写入一个文件，下次启动时重新加载
        save_to_disk(std::move(task));
    }
}

3.2 超时关闭

有些任务可能卡住了，一直不结束。这时候不能无限等下去。我建议加一个超时机制：

void shutdown_with_timeout(std::chrono::seconds timeout) {
    stop_accept_.store(true);
    stop_.store(true);
    cv_.notify_all();
    
    auto deadline = std::chrono::steady_clock::now() + timeout;
    
    for (auto& worker : workers_) {
        if (worker.joinable()) {
            auto remaining = deadline - std::chrono::steady_clock::now();
            if (remaining > std::chrono::seconds(0)) {
                worker.join();  // 正常等待
            } else {
                worker.detach();  // 超时，分离线程
                // 记录警告日志
                log_warning("Worker thread timed out, detached");
            }
        }
    }
}

四、整体架构图

下面这张图展示了线程池进阶版本的完整架构。你可以看到动态调整、优先级队列、优雅关闭三个模块如何协同工作。

五、避坑指南

最后，分享几个我踩过的坑：

• 线程数不要频繁调整：创建和销毁线程是有代价的。我建议至少间隔5秒做一次调整，避免抖动。
• 优先级队列不要无限增长：设置最大容量，超过后拒绝低优先级任务。否则内存会爆。
• 关闭时注意死锁：如果工作线程持有锁，而关闭线程也在等同一个锁，就会死锁。我习惯在关闭前先释放所有锁。
• 任务优先级不要太多：3-5个优先级足够了。太多优先级反而增加调度开销，得不偿失。

嗯，这一章的内容就到这里。线程池进阶其实还有很多细节，比如任务窃取、工作窃取、自适应调度等等。但掌握了动态调整、优先级队列和优雅关闭这三个核心能力，你的线程池已经能应对大部分生产场景了。