windows多线程demo-20240324

duqn

ThreadPool.h

1. #ifndef THREADPOOL_H
   
2. #define THREADPOOL_H
   
3. 
   
4. #include <vector>
   
5. #include <queue>
   
6. #include <thread>
   
7. #include <mutex>
   
8. #include <condition_variable>
   
9. #include <future>
   
10. #include <functional>
    
11. #include <stdexcept>
    
12. 
    
13. class ThreadPool {
    
14. public:
    
15.     ThreadPool(size_t threads) : stop(false) {
    
16.         for (size_t i = 0; i < threads; ++i)
    
17.             workers.emplace_back(
    
18.                 [this] {
    
19.                     for (;;) {
    
20.                         std::function<void()> task;
    
21. 
    
22.                         {
    
23.                             std::unique_lock<std::mutex> lock(this->queue_mutex);
    
24.                             this->condition.wait(lock,
    
25.                                 [this] { return this->stop || !this->tasks.empty(); });
    
26.                             if (this->stop && this->tasks.empty())
    
27.                                 return;
    
28.                             task = std::move(this->tasks.front());
    
29.                             this->tasks.pop();
    
30.                         }
    
31. 
    
32.                         task();
    
33.                     }
    
34.                 }
    
35.         );
    
36.     }
    
37. 
    
38.     ~ThreadPool() {
    
39.         {
    
40.             std::unique_lock<std::mutex> lock(queue_mutex);
    
41.             stop = true;
    
42.         }
    
43.         condition.notify_all();
    
44.         for (std::thread& worker : workers)
    
45.             worker.join();
    
46.     }
    
47. 
    
48.     template<class F, class... Args>
    
49.     auto enqueue(F&& f, Args&&... args)
    
50.         -> std::future<typename std::result_of<F(Args...)>::type> {
    
51.         using return_type = typename std::result_of<F(Args...)>::type;
    
52. 
    
53.         auto task = std::make_shared< std::packaged_task<return_type()> >(
    
54.             std::bind(std::forward<F>(f), std::forward<Args>(args)...)
    
55.         );
    
56. 
    
57.         std::future<return_type> res = task->get_future();
    
58.         {
    
59.             std::unique_lock<std::mutex> lock(queue_mutex);
    
60. 
    
61.             if (stop)
    
62.                 throw std::runtime_error("enqueue on stopped ThreadPool");
    
63. 
    
64.             tasks.emplace([task]() { (*task)(); });
    
65.         }
    
66.         condition.notify_one();
    
67.         return res;
    
68.     }
    
69. 
    
70. private:
    
71.     std::vector< std::thread > workers;
    
72.     std::queue< std::function<void()> > tasks;
    
73.     std::mutex queue_mutex;
    
74.     std::condition_variable condition;
    
75.     bool stop;
    
76. };
    
77. 
    
78. #endif
    

複製代碼

main.cpp

1. #include "ThreadPool.h"
   
2. #include <iostream>
   
3. #include <vector>
   
4. #include <chrono>
   
5. #include <random>
   
6. 
   
7. struct TaskInfo {
   
8.     int id;
   
9.     std::chrono::milliseconds start_time;
   
10.     std::chrono::milliseconds end_time;
    
11.     double duration() const {
    
12.         return std::chrono::duration_cast<std::chrono::microseconds>(end_time - start_time).count() / 1000.0;
    
13.     }
    
14. };
    
15. 
    
16. int main() {
    
17.     ThreadPool pool(10); // 使用10个线程的线程池
    
18. 
    
19.     std::vector<TaskInfo> taskInfos(10);
    
20.     std::vector<std::future<void>> results;
    
21. 
    
22.     auto start = std::chrono::high_resolution_clock::now();
    
23. 
    
24.     for (int i = 0; i < 10; ++i) {
    
25.         results.emplace_back(
    
26.             pool.enqueue([i, &taskInfos, start]() {
    
27.                 auto task_start = std::chrono::high_resolution_clock::now();
    
28. 
    
29.                 // 模拟工作负载
    
30.                 std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 假设每个任务需要100毫秒
    
31. 
    
32.                 auto task_end = std::chrono::high_resolution_clock::now();
    
33. 
    
34.                 taskInfos[i] = { i, std::chrono::duration_cast<std::chrono::milliseconds>(task_start - start),
    
35.                                 std::chrono::duration_cast<std::chrono::milliseconds>(task_end - start) };
    
36. 
    
37.                 // 此处可以添加打印任务完成信息的代码
    
38.                 std::cout << "Task" << i << " over!" << std::endl;
    
39.                 })
    
40.         );
    
41.     }
    
42. 
    
43.     // 确保所有任务已经完成
    
44.     for (auto& res : results) {
    
45.         res.get();
    
46.     }
    
47. 
    
48.     auto overall_end = std::chrono::high_resolution_clock::now();
    
49. 
    
50.     // 分析任务执行时间，找出最快和最慢的任务
    
51.     auto minmax = std::minmax_element(taskInfos.begin(), taskInfos.end(), [](const TaskInfo& a, const TaskInfo& b) {
    
52.         return a.duration() < b.duration();
    
53.         });
    
54. 
    
55.     // 输出最快和最慢的任务信息
    
56.     if (minmax.first != taskInfos.end() && minmax.second != taskInfos.end()) {
    
57.         std::cout << "most fast #" << minmax.first->id << ", used: " << minmax.first->duration() << " ms" << std::endl;
    
58.         std::cout << "most slow #" << minmax.second->id << ", used: " << minmax.second->duration() << " ms" << std::endl;
    
59.     }
    
60. 
    
61.     // 输出整体执行时间
    
62.     auto total_duration = std::chrono::duration_cast<std::chrono::milliseconds>(overall_end - start).count();
    
63.     std::cout << "all complete! total used: " << total_duration << " ms" << std::endl;
    
64. }

複製代碼