用 pthread 实现一个 C++ 线程池

这段封装里最关键的是 start_routine。线程真正跑起来以后，不是直接执行某个固定函数，而是回调到外面传进来的 std::function。后面线程池把自己的任务执行函数绑定进去，就能把线程对象和线程池逻辑接起来。

共享资源要保护

线程池里最核心的共享资源就是任务队列 _q。多个线程同时从队列里取任务，用户线程又可能往里塞任务，不加保护肯定会出问题。这里用一把互斥锁就够了，没必要把问题想复杂。

#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;
};

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

只有锁还不够。线程如果一直空转去抢锁、查队列，效率会很差，尤其是队列里根本没任务的时候。更合理的做法是：队列为空时让线程睡眠，等任务进来再唤醒它们。这就是条件变量的用处。

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

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

线程池本体

线程池内部其实就是三块东西：任务队列、线程数组、同步工具。线程数组里存的是线程对象，不是线程已经跑起来的结果；真正启动线程是在 Start() 里做的。

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;
};

这里的 std::vector<Thread> 只是把线程对象准备好，真正的线程还没创建。这个区别挺重要，不然很容易把'对象已存在'和'线程已运行'混成一件事。

构造阶段只做准备工作

构造函数里先把线程对象都塞进 _threads，每个线程绑定同一个执行入口：Routine()。线程池对象本身可以先建好，但不急着启动，这样更符合资源准备和真正执行分离的思路。

// 线程池
const static int defaultthreadnum = 3; // for debug

template <class T> class ThreadPool {
private:
    void Routine(const std::string &name) {
        while (true) {
            // 把任务从线程获取到线程私有！临界区 -> 私有的栈
            T t;
            {
                LockGuard lockguard(&_lock);
                while (QueueIsEmpty() && _is_running) {
                    _wait_thread_num++;
                    _cond.Wait(_lock);
                    _wait_thread_num--;
                }
                if (!_is_running && QueueIsEmpty()) {
                    LOG(LogLevel::INFO) << " 线程池退出 && 任务队列为空，" << name << " 退出";
                    break;
                }
                // 队列中一定有任务了!, 但是
                // 1. 线程池退出 -- 消耗历史
                // 2. 线程池没有退出 -- 正常工作
                t = _q.front();
                _q.pop();
            }
            t(); // 规定，未来的任务，必须这样处理！
            // for debug
            // LOG(LogLevel::DEBUG) << name << " handler task: " << t.Result2String();
        }
    }

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);
            _threads.emplace_back([this](const std::string &name) { 
                this->Routine(name); 
            }, name);
        }
        LOG(LogLevel::INFO) << "thread pool obj create success";
    }
};

这段里有个容易忽略的点：构造函数里的 for 循环不是在'创建线程'，而是在创建线程对象。对象一旦进了 vector，它当然已经存在了，但线程要等 Start() 被调用后才真正跑起来。

绑定 this 的做法也比较实用。Thread 只知道自己拿到一个可调用对象，至于这个可调用对象最终回到 ThreadPool::Routine()，那是线程池自己的事。这样层次分得清楚，不会把线程封装和任务调度搅在一起。

启动线程

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

停止线程池

void Stop() {
    if (!_is_running) return;
    _is_running = false;
    if (_wait_thread_num) _cond.NotifyAll();
    LOG(LogLevel::INFO) << "thread pool stop success";
}

这里直接把 _is_running 置为 false 再唤醒所有等待线程，思路很朴素，但也确实有点粗。更稳妥的退出方式一般要看线程当时在不在等任务、队列里还有没有历史任务。这个版本的逻辑是：

• 被唤醒时，如果队列空了，就退出；
• 被唤醒时，如果队列里还有任务，就先把任务处理完再退出；
• 没睡眠的线程就继续跑，直到把该处理的任务清掉。

第 3 种情况其实可以并到第 2 种里看，核心就是别把队列里的活半路扔掉。

回收线程

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

投递任务

void Enqueue(const T &t) {
    if (!_is_running) return;
    {
        LockGuard lockguard(&_lock);
        _q.push(t);
        if (_wait_thread_num > 0) _cond.NotifyOne();
    }
}

这套实现能跑起来，逻辑也比较直白。实际写线程池时，最容易出问题的不是'怎么把线程拉起来'，而是退出时怎么处理队列、怎么保证等待中的线程能被正确唤醒。线程池看着简单，真正麻烦的地方基本都在收尾。