Linux 环境下手写线程池设计与实现

同步原语

由于多个线程会并发访问任务队列，必须引入互斥锁保护共享资源。同时，为了避免线程在队列为空时不断轮询（忙等），需要使用条件变量让线程休眠，直到有新任务通知。

互斥锁封装 (Mutex)：

#pragma once
#include <pthread.h>
#include <iostream>
#include <unistd.h>
#include <string>

class Mutex {
public:
    Mutex() { pthread_mutex_init(&_lock, nullptr); }
    void Lock() { pthread_mutex_lock(&_lock); }
    void Unlock() { pthread_mutex_unlock(&_lock); }
    pthread_mutex_t *Get() { return &_lock; }
    ~Mutex() { pthread_mutex_destroy(&_lock); }
private:
    pthread_mutex_t _lock;
};

RAII 锁守卫 (LockGuard)：

使用 RAII 机制可以确保锁在作用域结束时自动释放，避免异常或提前返回导致的死锁：

class LockGuard {
public:
    LockGuard(Mutex *mutex) : _mutexp(mutex) { _mutexp->Lock(); }
    ~LockGuard() { _mutexp->Unlock(); }
private:
    Mutex *_mutexp;
};

条件变量 (Cond)：

用于线程间的通信与唤醒：

#pragma once
#include<pthread.h>
#include"Mutex.hpp"

class Cond {
public:
    Cond() { pthread_cond_init(&_cond,nullptr); }
    void Wait(Mutex &mutex) { pthread_cond_wait(&_cond,mutex.Get()); }
    void NotifyOne() { pthread_cond_signal(&_cond); }
    void NotifyAll() { pthread_cond_broadcast(&_cond); }
    ~Cond() { pthread_cond_destroy(&_cond); }
private:
    pthread_cond_t _cond;
};

线程池实现

整合上述组件，构建通用的模板化线程池。这里使用 std::vector<Thread> 存储工作线程，std::queue<T> 作为任务队列。

template <class T> class ThreadPool {
private:
    std::queue<T> _q;              // 任务队列
    std::vector<Thread> _threads;  // 工作线程集合
    int _threadnum;                // 线程数
    int _wait_thread_num;          // 当前等待任务的线程数
    Mutex _lock;                   // 互斥锁
    Cond _cond;                    // 条件变量
    bool _is_running;              // 运行标志
};

初始化与线程创建

构造函数负责分配资源，但此时线程尚未启动。我们利用 Lambda 表达式将线程池的任务执行方法绑定到每个工作线程中。

const static int defaultthreadnum = 3;

template <class T> class ThreadPool {
    // ... 成员变量 ...

public:
    ThreadPool(int threadnum = defaultthreadnum) 
        : _threadnum(threadnum), _is_running(false), _wait_thread_num(0) {
        for (int i = 0; i < _threadnum; i++) {
            std::string name = "thread-" + std::to_string(i + 1);
            // 使用 emplace_back 直接在 vector 内存中构造 Thread 对象
            _threads.emplace_back([this](const std::string &name) { 
                this->Routine(name); 
            }, name);
        }
        LOG(LogLevel::INFO) << "thread pool obj create success";
    }

    void Start() {
        if (_is_running) return;
        _is_running = true;
        for (auto &t : _threads) {
            t.Start();
        }
        LOG(LogLevel::INFO) << "thread pool running success";
    }
    // ... 其他方法 ...
};

任务调度与执行逻辑

这是线程池最核心的部分。工作线程进入 Routine 循环后，需要判断任务队列状态：

1. 队列为空且运行中：线程应进入休眠，等待新任务通知。
2. 队列为空且已停止：线程应退出循环。
3. 队列不为空：取出任务执行。

bool QueueIsEmpty() { return _q.empty(); }

void Routine(const std::string &name) {
    while (true) {
        T t;
        {
            LockGuard lockguard(&_lock);
            // 如果队列为空且线程池仍在运行，则等待
            while (QueueIsEmpty() && _is_running) {
                _wait_thread_num++;
                _cond.Wait(lockguard._mutexp->Get()); // 注意：Wait 内部会解锁
                _wait_thread_num--;
            }
            // 如果线程池已停止且队列为空，退出
            if (!_is_running && QueueIsEmpty()) {
                LOG(LogLevel::INFO) << " 线程池退出 && 任务队列为空，" << name << " 退出";
                break;
            }
            // 队列中有任务，取出
            if (!QueueIsEmpty()) {
                t = _q.front();
                _q.pop();
            }
        } // 锁在此处释放
        
        // 临界区外执行任务，避免阻塞其他线程入队
        if (t) t(); 
        LOG(LogLevel::DEBUG) << name << " handler task: " << t.Result2String();
    }
}

任务提交

当外部向线程池提交任务时，只需将任务推入队列并通知至少一个等待的线程即可。

void Enqueue(const T &t) {
    if (!_is_running) return;
    {
        LockGuard lockguard(&_lock);
        _q.push(t);
        if (_wait_thread_num > 0) _cond.NotifyOne(); // 唤醒一个等待线程
    }
}

优雅退出与回收

线程池停止时不能简单粗暴地调用 pthread_cancel，否则可能导致资源未释放或数据不一致。正确的做法是设置停止标志，并广播唤醒所有线程，让它们检查状态后自行退出。

void Stop() {
    if (!_is_running) return;
    _is_running = false;
    // 唤醒所有正在等待的线程，让它们检查_is_running 标志
    if (_wait_thread_num) _cond.NotifyAll();
    LOG(LogLevel::INFO) << "thread pool stop success";
}

void Wait() {
    for (auto &t : _threads) {
        t.Join();
    }
    LOG(LogLevel::INFO) << "thread pool wait success";
}

通过上述流程，我们实现了一个具备基本生命周期管理、任务调度及线程安全的 C++ 线程池。在实际工程中，还需考虑任务异常处理、最大队列长度限制等扩展功能。