线程池学习2

项目地址：https://github.com/vit-vit/ctpl

辅助队列

namespace ctpl { namespace detail { template <typename T> class Queue { public: bool push(T const& value) { std::unique_lock<std::mutex> lock(this->mutex); //def名为lock的std::unique_lock智能锁对象,锁住传入的 this->mutex 互斥锁 this->q.push(value); //通过this指针访问当前对象的私有成员q return true; } // deletes the retrieved element, do not use for non integral types // 这里的pop的参数是通过引用带回,"输出型参数", 传入一个空杯,在函数里加满水,我再拿回函数 bool pop(T& v) { std::unique_lock<std::mutex> lock(this->mutex); if (this->q.empty()) return false; v = this->q.front(); this->q.pop(); return true; } bool empty() { std::unique_lock<std::mutex> lock(this->mutex); return this->q.empty(); } private: std::queue<T> q; std::mutex mutex; }; }

首先封装标准库<queue>为类模版<Queue>，线程安全的队列实现，作为线程池的任务容器。通过this指针访问当前对象的私有成员q，所有操作都通过std::unique_lock加锁，保证多线程环境下的队列操作安全。

thread_pool类之私有成员

private: // deleted //这几行代码是 C++ 中禁用线程池对象的拷贝 / 移动语义的核心写法， // 目的是保证线程池这类 “资源独占型对象” 的安全，避免因拷贝 / 移动导致的线程管理混乱、资源泄漏或数据竞争 thread_pool(const thread_pool&);// = delete; thread_pool(thread_pool&&);// = delete; thread_pool& operator=(const thread_pool&);// = delete; thread_pool& operator=(thread_pool&&);// = delete; void set_thread(int i) { // 让当前创建的工作线程，和线程池容器共享同一个 “停止标志”，且保证标志的生命周期安全 flag std::shared_ptr<std::atomic<bool>> flag(this->flags[i]); // a copy of the shared ptr to the flag 拷贝构造函数 //Lambda 表达式的类型是 C++ 编译器匿名的、唯一的闭包类型—— 这个类型没有名字，你根本无法手动写出！ auto f = [this, i, flag/* a copy of the shared ptr to the flag */]() { std::atomic<bool>& _flag = *flag; std::function<void(int id)>* _f; bool isPop = this->q.pop(_f); while (true) { while (isPop) { // if there is anything in the queue //用智能指针包裹任务指针，自动释放内存 std::unique_ptr<std::function<void(int id)>> func(_f); // at return, delete the function even if an exception occurred //执行任务，传入线程索引i (*_f)(i); //检查是否收到停止符号 if (_flag) return; // the thread is wanted to stop, return even if the queue is not empty yet else isPop = this->q.pop(_f); //继续取下一个任务 } // the queue is empty here, wait for the next command // 步骤1：加锁——保护后续所有临界区操作 std::unique_lock<std::mutex> lock(this->mutex); // 步骤2：修改等待计数——必须在锁保护下 ++this->nWaiting; // 步骤3：wait 操作——自动解锁休眠 + 唤醒后重新加锁 (检查是否有新任务/池子销毁/自己线程停止) this->cv.wait(lock, [this, &_f, &isPop, &_flag]() { isPop = this->q.pop(_f); return isPop || this->isDone || _flag; }); // 步骤4：修改等待计数——依然在锁保护下, 线程被唤醒, 空闲线程的数目减1 --this->nWaiting; // 但如果发现唤醒后，不是因为有新任务，而是因为要让线程停工或者销毁池子，则真的退出 if (!isPop) return; // if the queue is empty and this->isDone == true or *flag then return } }; // 释放旧的std::thread, 接替新的std::thread this->threads[i].reset(new std::thread(f)); // compiler may not support std::make_unique() } void init() { this->nWaiting = 0; this->isStop = false; this->isDone = false; } std::vector<std::unique_ptr<std::thread>> threads; std::vector<std::shared_ptr<std::atomic<bool>>> flags; detail::Queue<std::function<void(int id)>*> q; //设计为void(int id)是为了把所有任务封装成 接收线程id 无返回值的函数，是pool和thread的通用协议 std::atomic<bool> isDone; std::atomic<bool> isStop; std::atomic<int> nWaiting; // how many threads are waiting std::mutex mutex; std::condition_variable cv; };

// 工作线程集合（智能指针管理，自动释放）

std::vector<std::unique_ptr<std::thread>> threads;

// 每个线程的停止标志（原子类型，线程安全）

std::vector<std::shared_ptr<std::atomic<bool>>> flags;

// 任务队列（存储函数对象指针）

detail::Queue<std::function<void(int id)>*> q; // 线程池状态标志（原子类型）

//设计为void(int id)是为把所有任务封装成 接收线程id无返回值的函数，是pool和thread的通用协议

std::atomic<bool> isDone; // 是否所有任务都执行完成

std::atomic<bool> isStop; // 是否强制停止线程池

std::atomic<int> nWaiting; // 等待中的线程数量

这里的set_thread是创建工作线程的核心函数，线程启动后先尝试从队列取任务执行，任务执行完后进入等待状态，直到有新任务或停止信号，等待时释放 CPU 资源，避免空转，支持随时停止单个线程（通过 flag 标志）。

while(true)里的逻辑，有任务，_f是从队列取出的任务指针（new出来的），用unique_ptr包裹后，不管是正常执行完还是中途退出，func析构时都会自动delete _f，彻底避免内存泄漏；然后执行任务；然后_flag默认是false表示线程未释放，所以会循环取下一个任务，然后执行。如果任务队列空了，isPop会为false，也就跳出了while循环，然后进入cv段的语句，去修改nWaiting

解锁，然后判断谓语，谓语为真唤醒后加锁，谓语为假继续休眠。有任务+池子未销毁+线程未停止会被唤醒。而如果唤醒后发现不是有任务，说明唤醒的原因是让线程停工或者池子销毁，这个时候则真的退出。再然后

std::thread是 C++ 标准库的线程类，构造时传入 “可调用对象”（这里是 Lambdaf）