C++11 实现线程池

template<typename T>
class SafeQueue {
private:
	std::queue<T> m_queue;
	std::mutex m_mutex; //访问互斥信号量
public:
	SafeQueue() {};
	SafeQueue(SafeQueue&& other) {};
	~SafeQueue() {};

	bool empty() {
		std::unique_lock<std::mutex> lock(m_mutex);
		return m_queue.empty();
	}


	int size() {
		std::unique_lock<std::mutex> lock(m_mutex);
		return m_queue.size();
	}

	void enqueue(T& t) {
		std::unique_lock<std::mutex> lock(m_mutex);
		m_queue.emplace(t);
	}

	bool dequeue(T& t) {
		std::unique_lock<std::mutex> lock(m_mutex);

		if (m_queue.empty()) {
			return false;
		}

		t = std::move(m_queue.front());
		m_queue.pop();
		return true;
	}
};

// 向线程池提交要执行的函数 ①
template <typename F, typename... Args>
auto submit(F&& f, Args &&...args)->std::future<decltype(f(args...))> {
    
    // 将函数和参数绑定 ②
    std::function<decltype(f(args...))()> func = std::bind(std::forward<F>(f), std::forward<Args>(args)...);

    // 封装函数 ③
    auto task_ptr = std::make_shared<std::packaged_task<decltype(f(args...))()>>(func);

    std::function<void()> wrapper_func = [task_ptr]() {
        (*task_ptr)();
    };

    // 队列通用安全封包函数，并压入安全队列
    m_queue.enqueue(wrapper_func);

    // 唤醒一个等待中的线程
    m_conditional_lock.notify_one();

    // 返回先前注册的任务指针
    return task_ptr->get_future();
}

template<typename R, typename T, typename U>
R add(T x, U y) {
 return x+y;
}

decltype(x+y) add(T x, U y)

template<typename T, typename U>
auto add2(T x, U y) -> decltype(x+y){
    return x + y;
}

class ThreadWorker // 内置线程工作类
{
private:
    int m_id; // 工作id
​
    ThreadPool *m_pool; // 所属线程池
public:
    // 构造函数
    ThreadWorker(ThreadPool *pool, const int id) : m_pool(pool), m_id(id)
    {
    }
​
    // 重载()操作
    void operator()()
    {
        std::function<void()> func; // 定义基础函数类func
​
        bool dequeued; // 是否正在取出队列中元素
        
        // 判断线程池是否关闭，没有关闭则从任务队列中循环提取任务
        while (!m_pool->m_shutdown)
        {
            {
                // 为线程环境加锁，互访问工作线程的休眠和唤醒
                std::unique_lock<std::mutex> lock(m_pool->m_conditional_mutex);
​
                // 如果任务队列为空，阻塞当前线程
                if (m_pool->m_queue.empty())
                {
                    m_pool->m_conditional_lock.wait(lock); // 等待条件变量通知，开启线程
                }
​
                // 取出任务队列中的元素
                dequeued = m_pool->m_queue.dequeue(func);
            }
​
            // 如果成功取出，执行工作函数
            if (dequeued)
                func();
        }
    }
};

#ifndef THREAD_POOL_H
#define THREAD_POOL_H

#include <condition_variable>
#include <functional>
#include <future>
#include <mutex>
#include <queue>

template<typename T>
class SafeQueue {
private:
	std::queue<T> m_queue;
	std::mutex m_mutex; //访问互斥信号量
public:
	SafeQueue() {};
	SafeQueue(SafeQueue&& other) {};
	~SafeQueue() {};

	bool empty() {
		std::unique_lock<std::mutex> lock(m_mutex);
		return m_queue.empty();
	}


	int size() {
		std::unique_lock<std::mutex> lock(m_mutex);
		return m_queue.size();
	}

	void enqueue(T& t) {
		std::unique_lock<std::mutex> lock(m_mutex);
		m_queue.emplace(t);
	}

	bool dequeue(T& t) {
		std::unique_lock<std::mutex> lock(m_mutex);

		if (m_queue.empty()) {
			return false;
		}

		t = std::move(m_queue.front());
		m_queue.pop();
		return true;
	}
};

class ThreadPool {
private:

	// 线程工作类
	class ThreadWorker {
		int m_id; //线程id

		ThreadPool* m_pool;//所属线程池
	public:
		ThreadWorker(ThreadPool* pool, const int id) :m_pool(pool), m_id(id)
		{	}

		void operator()() {
			std::function<void()> func;//定义基础函数类func
			bool dequeued;//是否正在取出队列元素

			while (!m_pool->m_shutdown) {
				{
					// 为线程环境加锁，互访问工作线程的休眠和唤醒
					std::unique_lock<std::mutex> lock(m_pool->m_conditional_mutex);
					
					// 如果任务队列为空，阻塞当前线程
					if (m_pool->m_queue.empty()) {
						m_pool->m_conditional_lock.wait(lock);
					}
					
					// 取出任务
					dequeued = m_pool->m_queue.dequeue(func);
				}

				// 如果成功取出，执行工作函数
				if (dequeued) {
					func();
				}
			}
		}
	};

private:
	bool m_shutdown;// 线程池是否关闭
	SafeQueue<std::function<void()>> m_queue;//任务队列
	std::vector<std::thread> m_threads;// 工作线程队列
	std::mutex m_conditional_mutex;// 线程休眠锁互斥变量
	std::condition_variable m_conditional_lock;// 线程环境锁，可以让线程处于休眠或者唤醒状态
public:
	ThreadPool(const int n_threads = 4) :
		m_threads(std::vector<std::thread>(n_threads)),
		m_shutdown(false) {
	}
	ThreadPool(const ThreadPool&) = delete;
	ThreadPool(ThreadPool&&) = delete;
	ThreadPool& operator=(const ThreadPool&) = delete;
	ThreadPool& operator=(ThreadPool&&) = delete;

	// 初始化线程池
	void init() {
		for (int i = 0; i < m_threads.size(); i++) {
			m_threads.at(i) = std::thread(ThreadWorker(this, i));
		}
	}

	// 等待线程完成当前工作，关闭线程池
	void shutdown() {
		m_shutdown = true;
		m_conditional_lock.notify_all();// 唤醒所有工作线程

		for (int i = 0; i < m_threads.size(); i++) {
			if (m_threads.at(i).joinable()) {
				m_threads.at(i).join();// 并入主线程
			}
		}
	}

	// 向线程池提交要执行的函数
	template <typename F, typename... Args>
	auto submit(F&& f, Args &&...args)->std::future<decltype(f(args...))> {
		
		// 将函数和参数绑定
		std::function<decltype(f(args...))()> func = std::bind(std::forward<F>(f), std::forward<Args>(args)...);

		// 封装函数，方便管理
		auto task_ptr = std::make_shared<std::packaged_task<decltype(f(args...))()>>(func);

		std::function<void()> wrapper_func = [task_ptr]() {
			(*task_ptr)();
		};

		// 队列通用安全封包函数，并压入安全队列
		m_queue.enqueue(wrapper_func);

		// 唤醒一个等待中的线程
		m_conditional_lock.notify_one();

		// 返回先前注册的任务指针
		return task_ptr->get_future();
	}
};

#endif

#include "main.h"

#include <string>
#include <regex>
#include <iostream>

#include "ThreadPool.h"
using namespace std;


void multiply(const int a, const int b) {
	std::this_thread::sleep_for(std::chrono::milliseconds(2000));
	cout << a << " * " << b << " = " << a * b << endl;
}

void multiply_output(int& out, const int a, const int b) {
	std::this_thread::sleep_for(std::chrono::milliseconds(2000));
	cout << a << " * " << b << " = " << a * b << endl;
}

int main() {
	ThreadPool pool(4);
	pool.init();

	for (int i = 0; i < 3; i++) {
		for (int j = 0; j < 10; j++) {
			pool.submit(multiply, i, j);
		}
	}
	auto f1=pool.submit(multiply, 5, 10);
	f1.get();

	//std::this_thread::sleep_for(std::chrono::milliseconds(2000));

	pool.shutdown();
	return 0;
}