C++ 简单线程池实现：基于 pthreads 与条件变量同步

一、设计思路

一个基础的线程池需要解决以下几个核心问题：

• 资源容器化：预先创建固定数量的工作线程，避免频繁创建销毁带来的开销。
• 任务管理：使用队列缓存待执行任务，支持动态添加。
• 线程同步：当队列为空时，工作线程应进入等待状态；有新任务加入时，需唤醒休眠线程。这通常通过互斥锁（Mutex）和条件变量（Condition Variable）或信号量（Semaphore）实现。

二、数据结构选择

为了平衡访问效率与实现复杂度，本示例采用以下结构：

• 线程组：使用数组存储线程 ID，便于管理与遍历。
• 任务队列：使用双向链表（std::list）存放任务指针。链表在头部插入和尾/头删除操作具有较高的性能优势，且易于实现 FIFO 顺序。

三、核心类设计

我们将线程池封装为一个模板类 ThreadPool<T>，其中 T 代表任务类型。这种泛型设计允许线程池处理任意类型的函数对象或可调用对象。

第 1 步：定义线程池接口

首先定义类的成员变量与基本方法，明确内部状态的管理权限。

template<typename T>
class ThreadPool {
public:
    // 构造函数：初始化线程数量与任务队列最大容量
    ThreadPool(int thread_num, int max_request);
    ~ThreadPool();

    // 将任务添加到队列尾部
    bool append(T* request);

private:
    // 工作线程入口函数（静态成员，用于 pthread_create）
    static void* worker(void* arg);
    
    // 实际执行逻辑：从队列获取并执行任务
    void run();

private:
    int m_thread_num;               // 工作线程数量
    int m_max_request;              // 任务队列最大长度
    pthread_t* m_threads;           // 存储线程 ID 的数组
    std::list<T*> m_queuework;     // 双端链表作为任务缓冲区
    Semaphore m_sem;                // 用于线程间通知（此处假定为已封装的信号量类）
    Mutex m_locker;                 // 互斥锁，保护任务队列的线程安全
};

第 2 步：实现构造与初始化

构造函数负责验证参数合法性，分配内存，并创建所有工作线程。注意：创建线程后建议立即将其设置为分离状态（detach），否则需要在适当时机回收线程资源以防止内存泄漏，或者在主线程析构前 join 所有线程。

template<typename T>
ThreadPool<T>::ThreadPool(int thread_num, int max_request)
    : m_thread_num(thread_num), m_max_request(max_request) {
    if (thread_num <= 0 || max_request <= 0) {
        throw std::exception("Invalid arguments for ThreadPool");
    }

    m_threads = new pthread_t[thread_num];
    if (!m_threads) throw std::bad_alloc();

    for (int i = 0; i < thread_num; ++i) {
        // 创建线程，入口函数为静态成员 worker
        if (pthread_create(&m_threads[i], NULL, worker, this) != 0) {
            delete[] m_threads;
            m_threads = nullptr;
            throw std::exception("Failed to create thread");
        }
        // 将线程设为 detached 状态，系统自动回收资源
        if (pthread_detach(m_threads[i]) != 0) {
            delete[] m_threads;
            m_threads = nullptr;
            throw std::exception("Failed to detach thread");
        }
    }
}

第 3 步：析构函数与安全清理

在实际生产环境中，线程池通常需要停止所有工作线程并清空任务队列。本简易示例主要关注内存清理，但在完整实现中应加入停止标志位（stopped flag）并通知所有等待的线程退出。

template<typename T>
ThreadPool<T>::~ThreadPool() {
    delete[] m_threads;
    m_threads = nullptr;
    // 注意：此处应确保所有任务已执行完毕或被丢弃
    // 且工作线程已正确终止
}

第 4 步：任务追加与同步机制

append 方法是生产者端的核心。它必须在加锁状态下检查队列长度，防止溢出，并在成功入队后发送信号唤醒工作线程。

template<typename T>
bool ThreadPool<T>::append(T* request) {
    m_locker.lock();
    
    // 如果队列已满，拒绝添加新任务
    if (m_queuework.size() > m_max_request) {
        m_locker.unlock();
        return false;
    }

    m_queuework.push_back(request);
    
    m_locker.unlock();
    
    // 唤醒一个正在等待的任务线程
    m_sem.post();
    return true;
}

第 5 步：工作线程的运行逻辑

这是“消费者”逻辑。线程启动后进入无限循环，等待信号量。一旦获得信号，即尝试获取任务执行。这里使用了非阻塞式判断配合忙轮询或再次等待的策略来处理伪唤醒（Spurious Wakeup）情况——虽然本例使用了信号量直接等待，但逻辑上仍需确保取出的任务非空。

template<typename T>
void* ThreadPool<T>::worker(void* arg) {
    ThreadPool* pool = reinterpret_cast<ThreadPool*>(arg);
    pool->run();
    return nullptr;
}

template<typename T>
void ThreadPool<T>::run() {
    while (true) {
        // 1. 等待任务信号
        m_sem.wait();
        
        // 2. 加锁访问共享队列
        m_locker.lock();
        
        if (m_queuework.empty()) {
            m_locker.unlock();
            continue; // 可能是虚假唤醒，继续等待
        }

        // 3. 取出任务
        T* request = m_queuework.front();
        m_queuework.pop_front();
        
        m_locker.unlock();

        // 4. 执行任务（注意：应在锁外执行，以缩短持有锁的时间）
        if (request) {
            request->process();
            delete request; // 假设任务由外部 new 创建，线程池负责释放
        }
    }
}

四、使用说明与注意事项

1. 任务类型要求：传入的任务类型 T 必须包含一个无参数的 process() 成员函数，或者您需要修改 run 方法来适配函数指针/std::function。
2. 生命周期管理：在线程池析构时，请务必确保没有未完成的任务仍在引用堆上的对象，否则可能导致悬挂指针或崩溃。建议在停止线程池后，先清空或等待队列完成。
3. 编译环境：该代码依赖 Posix Threads (pthreads)。在使用 g++ 编译时，务必链接 -lpthread 库：
   g++ -std=c++11 main.cpp -o pool -lpthread