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