C++11实现缓存线程池最佳实践

#ifndef SYNCQUEUE_HPP
#define SYNCQUEUE_HPP
#include<list>
#include<mutex>
#include<condition_variable>
#include<iostream>
using namespace::std;
template<class T>
class SynQueue
{
private:
    std::list<T> m_queue;                     // 任务存储容器
    mutable std::mutex m_mutex;               // 全局互斥锁，保护队列所有读写
    std::condition_variable m_notEmpty;       // 条件变量：队列非空，唤醒消费者
    std::condition_variable m_notFull;        // 条件变量：队列未满，唤醒生产者
    size_t m_waitTime;                        // 阻塞等待超时时间(秒)
    int m_maxSize;                            // 有界队列最大容量
    bool m_needStop;                          // 队列停止标记，用于优雅退出
    // 判断队列是否已满
    bool IsFull()const;
    // 判断队列是否为空
    bool IsEmpty()const;
    // 底层入队通用模板，完美转发统一处理左值、右值
    template<class F>
    int Add(F&& x);
public:
    // 构造：指定队列上限、等待超时时间
    SynQueue(int maxsize = 100, int waittime = 1);
    // 左值版本入队
    int Put(const T& x);
    // 右值版本入队
    int Put(T&& x);
    // 阻塞等待任务，仅做等待探测，不取出数据
    int notTask();
    // 批量取出全部任务，移动语义转移list
    void Take(std::list<T>& list);
    // 阻塞获取单个任务，支持超时
    int Take(T& t);
    // 优雅停止队列：等待任务消费完毕，唤醒所有阻塞线程
    void Stop();
    // 查询队列是否为空
    bool Empty()const;
    // 查询队列是否已满
    bool Full()const;
    // 获取当前任务数量
    size_t Size()const;
};
#endif // !SYNCQUEUE_HPP

template<class F>
int Add(F&& x)
{
	std::unique_lock<std::mutex> locker(m_mutex);
	// 等待队列有空位
	// 谓词返回 true 时停止等待（即：需要停止 OR 队列未满），wait_for 返回 false 表示超时
	if (!m_notFull.wait_for(locker, std::chrono::seconds(m_waitTime),
		[this] { return m_needStop || !IsFull(); })) 
	{
		std::cout << "task queue full, timeout return 1" << std::endl;
		return 1;		// 超时失败
	}
	// 检查是否收到停止信号
	if (m_needStop)
	{
		std::cout << "同步队列停止工作..." << std::endl;
		return 2;		// 停止状态
	}
	// 完美转发并入队
	m_queue.push_back(std::forward<F>(x));
	// 通知消费者
	m_notEmpty.notify_one();
	return 0;			// 成功
}

void Take(std::list<T>& list)
{
	// 获取互斥锁，保护共享队列 m_queue
	std::unique_lock<std::mutex> locker(m_mutex);
	// 循环等待：当系统未停止且队列为空时，阻塞当前线程
	while (!m_needStop && IsEmpty())
	{
		m_notEmpty.wait(locker);
	}
	// 退出检查：若是因为收到停止信号而跳出循环，则直接返回
	if (m_needStop) 
	{
		std::cout << "同步队列停止工作..." << std::endl;
		return;
	}
	// 批量提取：使用移动语义将整个队列内容转移给外部 list，效率极高（O(1)）
	list = std::move(m_queue);
	// 通知生产者：队列已清空，唤醒一个正在等待“非满”条件的生产线程
	m_notFull.notify_one();
}
int Take(T& t)
{
	std::unique_lock<std::mutex> locker(m_mutex);
	// 带超时的条件等待
	// 谓词 [this] { return !m_needStop && !IsEmpty(); } ，若返回 false (超时)，进入 if 分支
	if (!m_notEmpty.wait_for(locker, std::chrono::seconds(m_waitTime),
		[this] { return !m_needStop && !IsEmpty(); }))
	{
		return 1;       // 等待超时，队列仍为空
	}
	if (m_needStop)
	{
		std::cout << "同步队列停止工作..." << std::endl;
		return 2;       // 系统已请求停止
	}
	// 提取队首元素
	t = m_queue.front();
	m_queue.pop_front();
	m_notFull.notify_one();
	return 0;           // 成功获取任务
}

void Stop()
{
	std::unique_lock<std::mutex> locker(m_mutex);
	// 等待队列清空：防止在还有任务未处理时强行停止，确保数据完整性
	while (!IsEmpty())
	{
		m_notFull.wait(locker);
	}
	// 设置停止标志：通知所有工作线程准备退出
	m_needStop = true;
	// 广播通知：唤醒所有阻塞在 m_notEmpty（等待任务）和 m_notFull（等待空间）的线程
	m_notEmpty.notify_all();
	m_notFull.notify_all();
}

#ifndef CACHEDTHREADPOOL_HPP
#define CACHEDTHREADPOOL_HPP
#include"SynQueue2.hpp"
#include<map>
#include<future>
#include<functional>
#include<unordered_map>
using namespace std;
int MaxTaskCount = 2;
const int KeepAliveTime = 10;
class CachedThreadPool
{
public:
	using Task = std::function<void(void)>;		
private:
	std::unordered_map<std::thread::id, std::shared_ptr<std::thread>> m_threadgroup;	//线程组
	int m_coreThreadSize;					//核心线程下限
	int m_maxThreadSize;					//最大线程上限
	std::atomic_int m_idleThreadSize;		//空闲线程计数
	std::atomic_int m_curThreadSize;		//当前总线程数量
	mutable std::mutex m_mutex;				//容器互斥锁
	SynQueue<Task> m_queue;					//任务同步队列
	std::atomic_bool m_running;				//线程池运行标记
	std::once_flag m_flag;					//保证Stop仅执行一次
	void Start(int numthreads);
	void RunInThread();
	void StopThreadGroup();
public:
	CachedThreadPool(int initNumThreads=8,int taskPoolSize=MaxTaskCount);
	~CachedThreadPool();
	void Stop();
	//提交无返回值任务
	template<class Func,class... Args>
	void execute(Func&& func, Args&&... args);
	//提交带返回值任务
	template<class Func, class... Args>
	auto submit(Func&& func, Args&&... args)
		-> std::future<decltype(func(args...))> ;
};
#endif

void RunInThread()
{
	auto tid = std::this_thread::get_id();
	//记录线程空闲起始时间
	auto startTime = std::chrono::high_resolution_clock().now();
	while (m_running)
	{
		Task task;
		if (m_queue.Size() == 0 && m_queue.notTask())
		{
			auto now = std::chrono::high_resolution_clock().now();
			auto intervalTime = std::chrono::duration_cast<std::chrono::seconds>(now - startTime);
			std::lock_guard<std::mutex> lock(m_mutex);
			//空闲超时 && 当前线程数大于核心线程，销毁临时线程
			if (intervalTime.count() >= KeepAliveTime && m_curThreadSize > m_coreThreadSize)
			{
				m_threadgroup.find(tid)->second->detach();
				m_threadgroup.erase(tid);
				m_curThreadSize--;
				m_idleThreadSize--;
				cout << "空闲线程销毁 " << m_curThreadSize << " " << m_coreThreadSize << endl;
				return;
			}
			//阻塞获取任务并执行
			if (!m_queue.Take(task) && m_running)
			{
				m_idleThreadSize--;
				task();
				m_idleThreadSize++;
				//任务执行完毕，重置空闲计时
				startTime = std::chrono::high_resolution_clock().now();
			}
		}
	}
}

template<class Func, class... Args>
auto submit(Func&& func, Args&&... args)
	-> std::future<decltype(func(args...))> 
{
	using RetType = decltype(func(args...));
	//封装任务，支持获取返回值
	auto task = std::make_shared<std::packaged_task<RetType()>>(
		std::bind(std::forward<Func>(func),std::forward<Args>(args)...)
	);
	std::future<RetType> result = task->get_future();
	//尝试将任务放入同步队列
	if (m_queue.Put([task]() {(*task)();})!=0)
	{
		std::cout << "调用者运行策略" << std::endl;
		(*task)();
	}
	//无空闲线程，且未达到最大线程限制，新建线程扩容
	if (m_idleThreadSize<=0&&m_curThreadSize<m_maxThreadSize)
	{
		std::lock_guard<std::mutex> lock(m_mutex);
		auto tha = std::make_shared<std::thread>(
			std::thread(&CachedThreadPool::RunInThread,this)
		);
		std::thread::id tid = tha->get_id();
		tha->detach();
		m_threadgroup.emplace(tid, std::move(tha));
		m_idleThreadSize++;
		m_curThreadSize++;
	}
	return result;
}

void StopThreadGroup()
{
	m_queue.Stop();		//停止同步队列，禁止新任务入队
	m_running = false;	//修改线程池运行标志
	//等待所有工作线程执行结束
	for (auto thread : m_threadgroup) 
	{
		thread.second->join();
	}
	m_threadgroup.clear();
}

//无返回值测试任务
void func(int index)
{
    static int num = 0;
    //静态变量验证多线程并发访问
    cout << "func_" << index << " num: " << ++num << endl;
}
//带返回值测试任务
int add(int a, int b)
{
    return a + b;
}
int main()
{
    //创建可缓存动态线程池
    CachedThreadPool mypool;
    //循环批量提交1000个任务
    for (int i = 0; i < 1000; ++i)
    {
        if (i % 2 == 0)
        {
            //偶数：提交带返回值任务，通过future接收结果
            //auto pa = mypool.submit(add, i, i + 1);
            auto pa = mypool.submit([=]() { return add(i, i + 1); });
            //阻塞等待任务完成，获取返回值并打印
            cout << pa.get() << endl;
        }
        else
        {
            //奇数：提交无需返回值的任务
            mypool.execute(func, i);
        }
    }
    return 0;
}

//创建核心线程数量为2的动态线程池
CachedThreadPool pool(2);
//耗时任务，模拟长时间业务处理
int add(int a, int b, int s)
{
    //任务休眠s秒，占用工作线程
    std::this_thread::sleep_for(std::chrono::seconds(s));
    int c = a + b;
    cout << "add begin ..." << endl;
    return c;
}
//在线程池提交任务
void add_a()
{
    //向线程池提交耗时4s任务
    auto r = pool.submit(add, 10, 20, 4);
    cout << "add_a: " << r.get() << endl;
}
void add_b()
{
    //向线程池提交耗时6s任务
    auto r = pool.submit(add, 20, 30, 6);
    cout << "add_b: " << r.get() << endl;
}
void add_c()
{
    //向线程池提交耗时1s任务
    auto r = pool.submit(add, 30, 40, 1);
    cout << "add_c: " << r.get() << endl;
}
void add_d()
{
    //向线程池提交耗时9s任务
    auto r = pool.submit(add, 10, 40, 9);
    cout << "add_d: " << r.get() << endl;
}
int main()
{
    //开启4个独立线程并发向线程池投递任务，制造任务突发压力
    std::thread tha(add_a);
    std::thread thb(add_b);
    std::thread thc(add_c);
    std::thread thd(add_d);
    tha.join();
    thb.join();
    thc.join();
    thd.join();
    //休眠20s，等待空闲线程触发超时回收逻辑
    std::this_thread::sleep_for(std::chrono::seconds(20));
    //再次发起新一轮任务
    std::thread the(add_a);
    std::thread thf(add_b);
    the.join();
    thf.join();
    return 0;
}