半同步半异步线程池

半同步半异步线程池分为三层：

• 同步服务层 —— 处理来自上层的任务请求，将它们加入到排队层中等待处理。

• 同步排队层 —— 实际上是一个“同步队列”，允许多线程添加/取出任务，并保证线程安全。

• 异步服务层 —— 从排队层中取出任务，多线程并发处理排队层中的任务。

不想码字，想看的凑活着看吧！

首先，我们来实现一个 同步队列 的模板：

#pragma once#include<iostream>
#include<thread>
#include<mutex>
#include<list>template <typename T>
class Sync_Queue
{
public:Sync_Queue(int size) : max_size(size), _stop(false){}void push(T&& x)    // 添加任务{std::unique_lock<std::mutex> lock(_mutex);_notFull.wait(lock, [this] { return NotFull() || _stop; });  // 若满足其中任一条件，则继续执行if (_stop)return;_queue.push_back(std::forward<T>(x));_notEmpty.notify_one();}void pop(std::list<T>& list)    // 取出任务{std::unique_lock<std::mutex> lock(_mutex);_notEmpty.wait(lock, [this] { return NotEmpty() || _stop; });if (_stop)return;list = std::move(_queue);_notFull.notify_one();}void stop()    // 停止队列{{std::lock_guard<std::mutex> lock(_mutex);    // 先锁住, 再将 _stop 标志设置为 true_stop = true;}_notFull.notify_all();   // 在 lock_guard 外面 notify, 被唤醒的线程不需要等待 lock_guard 释放锁 _notEmpty.notify_all();}bool Empty(){std::lock_guard<std::mutex> lock(_mutex);return _queue.empty();}bool Full(){std::lock_guard<std::mutex> lock(_mutex);return _queue.size() == max_size;}size_t size(){std::lock_guard<std::mutex> lock(_mutex);return _queue.size();}private:bool NotFull() const{bool notfull = _queue.size() < max_size;if (!notfull) std::cout << "Sync_Queue is full, waiting..." << std::endl;return notfull;}bool NotEmpty() const{bool notempty = !_queue.empty();if (!notempty) std::cout << "Sync_Queue is empty, waiting..." << std::endl;return notempty;}private:std::list<T> _queue;std::mutex _mutex;std::condition_variable _notEmpty;    // 非空的条件变量std::condition_variable _notFull;     // 未满的条件变量int max_size;bool _stop;
};

现在，我们再来实现 线程池：

// ThreadPool.h
#pragma once#include "Sync_Queue.h"
#include <atomic>
#include <memory>
#include <functional>using Task = std::function<void()>;  // 任务类型为一个 “可调用对象”const int MaxTaskCount = 100;class ThreadPool {
public:ThreadPool(int thread_num = std::thread::hardware_concurrency())  // 默认创建 CPU 核数的线程: _queue(MaxTaskCount), thread_stop(false){start(thread_num);}~ThreadPool(){stop();}void stop(){std::call_once(_flag, [this] { StopThreadPool(); });  // 确保多线程下只调用一次}void add_task(Task&& task)      // 添加任务{_queue.push(std::forward<Task>(task));}private:void start(int thread_num)     // 创建 thread_num 数量的线程{for (int i = 0; i < thread_num; ++i){thread_group.push_back(std::make_shared<std::thread>(&ThreadPool::RunInThread, this));}}void RunInThread(){while (!thread_stop) {std::list<Task> list;_queue.pop(list);      // 取任务; 若消息队列为空，则阻塞for (auto& task : list){if (thread_stop)return;task();     // 执行任务}}}void StopThreadPool(){_queue.stop();thread_stop = true;for (auto thread : thread_group) {if (thread->joinable())thread->join();}thread_group.clear();}private:Sync_Queue<Task> _queue;   // 同步队列std::list<std::shared_ptr<std::thread>> thread_group;   // 线程组std::atomic_bool thread_stop;std::once_flag _flag;
};

测试代码：

#include "ThreadPool.h"
#include <chrono>void test()
{ThreadPool pool(3);std::thread t1([&pool] {for (int i = 0; i < 10; ++i){auto id = std::this_thread::get_id();pool.add_task(std::move([id] {std::cout << "thread id is " << id << std::endl;}));}});std::this_thread::sleep_for(std::chrono::seconds(2));getchar();t1.join();
}int main()
{test();return 0;
}

输出如下：