c++11多线程-从零基础到手撕线程池

#include<iostream> #include<thread> #include<mutex> #include<string> #include<condition_variable> #include<queue> #include<vector> #include<functional> class ThreadPool { public: ThreadPool(int numThreads) : stop(false) { for (int i = 0; i < numThreads; i++) { threads.emplace_back([this]() { while (1) { std::unique_lock<std::mutex> lock(mtx); condition.wait(lock, [this]() { return !tasks.empty() || stop; }); if (stop && tasks.empty()) { return; } std::function<void()> task(std::move(tasks.front())); tasks.pop(); lock.unlock(); task(); } }); } } ~ThreadPool() { { std::unique_lock<std::mutex> lock(mtx); stop = true; } condition.notify_all(); for (auto& t : threads) { t.join(); } } template<class T, class... Args> void enqueue(T&& t, Args&&... args) { std::function<void()> task = std::bind(std::forward<T>(t), std::forward<Args>(args)...); { std::unique_lock<std::mutex> lock(mtx); tasks.emplace(std::move(task)); } condition.notify_one(); } private: std::vector<std::thread> threads; std::queue<std::function<void()>> tasks; std::mutex mtx; std::condition_variable condition; bool stop; }; class ThreadPoolSingleton { public: ThreadPoolSingleton(const ThreadPoolSingleton&) = delete; ThreadPoolSingleton& operator=(const ThreadPoolSingleton&) = delete; static ThreadPoolSingleton& getInstance() { static ThreadPoolSingleton instance; return instance; } template<class T, class... Args> void enqueue(T&& t, Args&&... args) { std::function<void()> task = std::bind(std::forward<T>(t), std::forward<Args>(args)...); { std::unique_lock<std::mutex> lock(mtx); tasks.emplace(std::move(task)); } condition.notify_one(); } private: ThreadPoolSingleton() : stop(false) { int numThreads = std::thread::hardware_concurrency(); // 硬件支持的线程数 for (int i = 0; i < numThreads; i++) { threads.emplace_back([this]() { while (1) { std::unique_lock<std::mutex> lock(mtx); condition.wait(lock, [this]() { return !tasks.empty() || stop; }); if (stop && tasks.empty()) { return; } std::function<void()> task(std::move(tasks.front())); tasks.pop(); lock.unlock(); task(); } }); } } ~ThreadPoolSingleton() { { std::unique_lock<std::mutex> lock(mtx); stop = true; } condition.notify_all(); for (auto& t : threads) { t.join(); } } std::vector<std::thread> threads; std::queue<std::function<void()>> tasks; std::mutex mtx; std::condition_variable condition; bool stop; }; int main() { //ThreadPool pool(4); //for (int i = 0; i < 10; i++) { // pool.enqueue([i]() { // std::cout << "任务：" << i << "正在运行" << std::endl; // std::this_thread::sleep_for(std::chrono::seconds(1)); // std::cout << "任务：" << i << "结束运行" << std::endl; // }); //} ThreadPoolSingleton::getInstance(); for (int i = 0; i < 10; i++) { ThreadPoolSingleton::getInstance().enqueue([i]() { std::cout << "任务：" << i << "正在运行" << std::endl; std::this_thread::sleep_for(std::chrono::seconds(1)); std::cout << "任务：" << i << "结束运行" << std::endl; }); } return 0; }