高级线程管理_第九章_《C++并发编程实战》笔记

1. 线程池（Thread Pool）

核心目的：避免频繁创建/烧毁线程，复用固定数量的线程处理任务队列。
1.1 线程池布局要素

• 任务队列：存储待执行的任务（函数对象）
  
• 工作线程集合：执行任务的线程
  
• 同步机制：互斥锁（std::mutex）、条件变量（std::condition_variable）
  
• 克制标志：安全停止池的开关（std::atomic）
  

1.2 线程池实现步调

• 步调1：定义任务队列和工作线程
  
• 步调2：初始化工作线程，循环等候任务
  
• 步调3：提交任务到队列，唤醒线程
  
• 步调4：安全关闭池，等候剩余任务完成
  代码示例:
  

1. #include <iostream>
2. #include <vector>
3. #include <queue>
4. #include <thread>
5. #include <mutex>
6. #include <condition_variable>
7. #include <future>
8. #include <functional>
10. class ThreadPool {
11. public:
12.     explicit ThreadPool(size_t num_threads) : stop(false) {
13.         for(size_t i = 0; i < num_threads; ++i) {
14.             workers.emplace_back([this] {
15.                 while(true) {
16.                     std::function<void()> task;
17.                     {
18.                         std::unique_lock<std::mutex> lock(queue_mutex);
19.                         condition.wait(lock, [this] { return stop || !tasks.empty(); });
20.                         if (stop && tasks.empty()) return;
21.                         task = std::move(tasks.front());
22.                         tasks.pop();
23.                     }
24.                     task();
25.                 }
26.             });
27.         }
28.     }
30.     template<class F, class... Args>
31.     auto enqueue(F&& f, Args&&... args) -> std::future<decltype(f(args...))> {
32.         using return_type = decltype(f(args...));
33.         auto task = std::make_shared<std::packaged_task<return_type()>>(
34.             std::bind(std::forward<F>(f), std::forward<Args>(args)...)
35.         );
36.         std::future<return_type> res = task->get_future();
37.         {
38.             std::unique_lock<std::mutex> lock(queue_mutex);
39.             if(stop) throw std::runtime_error("enqueue on stopped ThreadPool");
40.             tasks.emplace([task]() { (*task)(); });
41.         }
42.         condition.notify_one();
43.         return res;
44.     }
46.     ~ThreadPool() {
47.         {
48.             std::unique_lock<std::mutex> lock(queue_mutex);
49.             stop = true;
50.         }
51.         condition.notify_all();
52.         for(std::thread &worker : workers)
53.             worker.join();
54.     }
56. private:
57.     std::vector<std::thread> workers;
58.     std::queue<std::function<void()>> tasks;
59.     std::mutex queue_mutex;
60.     std::condition_variable condition;
61.     std::atomic<bool> stop;
62. };
64. // 测试用例
65. int main() {
66.     ThreadPool pool(4);
67.     std::vector<std::future<int>> results;
69.     for(int i = 0; i < 8; ++i) {
70.         results.emplace_back(pool.enqueue([i] {
71.             std::cout << "Task " << i << " executed by thread "
72.                       << std::this_thread::get_id() << std::endl;
73.             std::this_thread::sleep_for(std::chrono::seconds(1));
74.             return i * i;
75.         }));
76.     }
78.     for(auto&& result : results)
79.         std::cout << "Result: " << result.get() << std::endl;
81.     return 0;
82. }

复制代码

2. 线程停止（Interruptible Threads）

核心目标：安全地请求线程停止，而非强制停止（避免资源泄漏）。
2.1 停止机制实现

• 检查点：线程定期检查停止标志
  
• 原子标志：std::atomic 作为停止信号
  
• 异常通报：通过抛出特定异常跳出任务
  代码:
  

1. #include <iostream>
2. #include <thread>
3. #include <atomic>
4. #include <stdexcept>
5. #include <mutex>
7. // 线程安全输出的互斥锁
8. std::mutex cout_mutex;
10. class InterruptFlag {
11. public:
12.     void set() { flag.store(true); }
13.     bool is_set() const { return flag.load(); }
14. private:
15.     std::atomic<bool> flag{false};
16. };
18. class InterruptibleThread {
19. public:
20.     template<typename FunctionType>
21.     InterruptibleThread(FunctionType f, InterruptFlag& flag)
22.         : flag(flag), thread([this, f, &flag] {
23.             try {
24.                 f(flag);
25.             } catch (...) {
26.                 std::lock_guard<std::mutex> lock(cout_mutex);
27.                 std::cerr << "Thread interrupted by an unknown exception." << std::endl;
28.             }
29.         })
30.     {}
32.     void join() { thread.join(); }
33.     void interrupt() { flag.set(); }
35. private:
36.     InterruptFlag& flag;
37.     std::thread thread;
38. };
40. // 测试用例
41. void task(InterruptFlag& flag) {
42.     while (!flag.is_set()) {
43.         {
44.             std::lock_guard<std::mutex> lock(cout_mutex);
45.             std::cout << "Working..." << std::endl;
46.         }
47.         std::this_thread::sleep_for(std::chrono::milliseconds(500));
48.     }
49.     throw std::runtime_error("Exit by interrupt");
50. }
52. int main() {
53.     InterruptFlag flag;
54.     InterruptibleThread t(task, flag);
55.    
56.     std::this_thread::sleep_for(std::chrono::seconds(2));
57.     {
58.         std::lock_guard<std::mutex> lock(cout_mutex);
59.         std::cout << "Sending interrupt..." << std::endl;
60.     }
61.     t.interrupt();
62.     t.join();
64.     return 0;
65. }
67. /*
68. - 协作式中断：线程在安全点检查标志，决定是否退出。
69. - 异常处理：通过抛出异常跳出深层嵌套调用栈。
70. - 资源释放：确保线程退出前正确释放持有资源。
71. */

复制代码

多选题

1.（多选）在线程池计划中，以下哪种情况会导致任务无法被正确执行？
A. 工作线程在pop_task时未正确处理条件变量的虚伪唤醒
B. 使用shared_ptr生存任务队列的锁，导致死锁
C. 提交任务时未检查线程池是否已关闭，直接放入队列
D. 任务队列的互斥锁粒度计划过大，影响并发性能
E. 在析构线程池时未优雅关闭所有线程
2.（多选）关于线程停止的正确计划，以下描述错误的是？
A. 可以通过atomic标志位实现线程的强制停止
B. C++标准库中的std::thread原生支持停止操作
C. 线程池中的任务应包含停止检查点以实现协作式停止
D. 停止一个阻塞在I/O操作上的线程可能导致资源泄漏
E. 使用条件变量等候时应联合谓词避免无法唤醒的阻塞
3.（多选）在线程池的任务调度中，以下哪些操作必要同步？
A. 检查任务队列是否为空
B. 多个工作线程同时从队列头部提取任务
C. 提交新任务到队列尾部
D. 调整线程池中的线程数量
E. 修改线程池的运行状态标志
4.（多选）线程池的优雅关闭应包含哪些步调？
A. 克制接受新任务
B. 等候所有已提交任务执行完成
C. 强制停止所有阻塞的工作线程
D. 通过条件变量通知所有工作线程退出循环
E. 重置任务队列中的未完成任务
5.（多选）以下哪些计划可能导致线程池的性能瓶颈？
A. 使用无锁队列实现任务提交
B. 工作线程频繁竞争同一互斥锁访问队列
C. 动态调整线程数量时未考虑负载平衡
D. 任务分发接纳简单的轮询策略
E. 使用线程当地存储缓存高频访问的数据
多选题答案

• 答案：A、C、E
  

剖析：
A. 工作线程在弹出任务时未处理条件变量的虚伪唤醒，可能导致线程错误地尝试处理空队列中的任务，从而崩溃或逻辑错误。
C. 提交任务时不检查线程池是否关闭，可能导致任务被添加到已关闭的队列中，无法执行或引发异常。
E. 析构线程池时未优雅关闭线程，可能导致资源泄漏或线程访问已烧毁的数据。
B错误：shared_ptr生存锁不会直接导致死锁，锁本身计划或使用顺序错误才会导致死锁。shared_ptr管理锁可能导致所有权混乱，但题目中“导致死锁”的描述不准确。
D错误：互斥锁粒度过大会降低并发性能，但不会直接导致任务无法执行。

• 答案：A、B
  

剖析：
A错误：atomic标志位无法强制停止阻塞中的线程（如等候条件变量或I/O），此为协作式停止，选项描述错误。
B错误：C++标准库的std::thread不原生支持停止，需通过标志位实现协作式停止。
C正确：任务应包含停止检查点。
D正确：停止阻塞在I/O的线程可能引发资源泄漏。
E正确：条件变量需联合谓词避免虚伪唤醒。

• 答案：A、B、C、D、E
  

剖析：
A. 检查队列是否为空需同步，以避免数据竞争。
B. 多个线程提取任务需同步，防止并发修改队列。
C. 提交任务到队列尾部需同步，保证队列的线程安全。
D. 调整线程数量需同步，避免中间状态不一致。
E. 修改运行状态标志需同步，确保可见性与原子性。

• 答案：A、B、D
  

剖析：
A. 克制接受新任务是优雅关闭的第一步。
B. 等候已提交任务完成保证任务全部处理。
D. 通过条件变量通知线程退出循环是关键步调。
C错误：强制停止线程是暴力方式，不优雅。
E错误：重置未完成任务非必要，取决于计划。

• 答案：B、C、D
  

剖析：
B. 频繁竞争同一锁（如任务队列锁）是范例性能瓶颈。
C. 动态调整线程数未负载平衡，可能导致资源浪费或竞争。
D. 轮询策略可能导致负载不均，影响性能。
A正确：无锁队列可提升性能。
E正确：线程当地存储优化高频数据访问，反而是优化点。
计划题目

• 底子线程池实现
  题目要求：计划一个固定巨细的线程池，支持提交无参数任务。实现任务队列、工作线程启动和任务分发逻辑，并在main函数中测试提交10个任务。
  
• 支持任务优先级的线程池
  题目要求：扩展线程池，实现任务优先级（高/低）。高优先级任务优先执行。在main函数中测试混淆优先级任务的执行顺序。
  
• 线程池中的任务返回值
  题目要求：实现线程池支持任务返回Future。使用std::packaged_task封装任务，允许调用者获取计算结果。
  
• 动态调整线程池巨细
  题目要求：允许运行时动态增加或镌汰工作线程数量。需处理线程的启动和退出同步问题。
  
• 线程安全的停止机制
  题目要求：为线程池中的任务添加协作式停止支持，允许外部请求停止长时间运行的任务。
  

计划题目答案

1. // 1.
2. #include <iostream>
3. #include <vector>
4. #include <thread>
5. #include <queue>
6. #include <mutex>
7. #include <condition_variable>
8. #include <functional>
9. #include <atomic>
11. class ThreadPool {
12. public:
13.     explicit ThreadPool(size_t num_threads) : stop(false) {
14.         for (size_t i = 0; i < num_threads; ++i) {
15.             workers.emplace_back([this] {
16.                 while (true) {
17.                     std::function<void()> task;
18.                     {
19.                         std::unique_lock<std::mutex> lock(queue_mutex);
20.                         condition.wait(lock, [this] { return stop || !tasks.empty(); });
21.                         if (stop && tasks.empty()) return;
22.                         task = std::move(tasks.front());
23.                         tasks.pop();
24.                     }
25.                     task();
26.                 }
27.             });
28.         }
29.     }
31.     template<typename F>
32.     void enqueue(F&& f) {
33.         {
34.             std::unique_lock<std::mutex> lock(queue_mutex);
35.             tasks.emplace(std::forward<F>(f));
36.         }
37.         condition.notify_one();
38.     }
40.     ~ThreadPool() {
41.         {
42.             std::unique_lock<std::mutex> lock(queue_mutex);
43.             stop = true;
44.         }
45.         condition.notify_all();
46.         for (auto& worker : workers)
47.             worker.join();
48.     }
50. private:
51.     std::vector<std::thread> workers;
52.     std::queue<std::function<void()>> tasks;
53.     std::mutex queue_mutex;
54.     std::condition_variable condition;
55.     std::atomic<bool> stop;
56. };
58. // 测试代码
59. int main() {
60.     ThreadPool pool(4);
61.     for (int i = 0; i < 10; ++i) {
62.         pool.enqueue([i] {
63.             std::this_thread::sleep_for(std::chrono::milliseconds(100));
64.             std::cout << "Task " << i << " done by thread "
65.                       << std::this_thread::get_id() << std::endl;
66.         });
67.     }
68.     std::this_thread::sleep_for(std::chrono::seconds(1)); // 等待任务完成
69.     return 0;
70. }
71. // 2.
72. #include <iostream>
73. #include <vector>
74. #include <queue>
75. #include <thread>
76. #include <mutex>
77. #include <condition_variable>
78. #include <functional>
79. #include <atomic>
81. class ThreadPool {
82. private:
83.     struct Task {
84.         std::function<void()> func;
85.         int priority; // 0:低, 1:高
86.         bool operator<(const Task& other) const { return priority < other.priority; }
87.     };
89.     std::vector<std::thread> workers;
90.     std::priority_queue<Task> tasks;
91.     std::mutex queue_mutex;
92.     std::condition_variable condition;
93.     std::atomic<bool> stop{ false };
95.     void worker() {
96.         while (true) {
97.             std::function<void()> task;
98.             {
99.                 std::unique_lock<std::mutex> lock(this->queue_mutex);
100.                 this->condition.wait(lock, [this] { return this->stop ||!this->tasks.empty(); });
101.                 if (this->stop && this->tasks.empty())
102.                     return;
103.                 task = std::move(this->tasks.top().func);
104.                 this->tasks.pop();
105.             }
106.             task();
107.         }
108.     }
110. public:
111.     ThreadPool(size_t numThreads) {
112.         for (size_t i = 0; i < numThreads; ++i) {
113.             workers.emplace_back([this] { this->worker(); });
114.         }
115.     }
117.     void enqueue(std::function<void()> f, int priority) {
118.         std::unique_lock<std::mutex> lock(queue_mutex);
119.         tasks.push({ std::move(f), priority });
120.         condition.notify_one();
121.     }
123.     ~ThreadPool() {
124.         {
125.             std::unique_lock<std::mutex> lock(queue_mutex);
126.             stop = true;
127.         }
128.         condition.notify_all();
129.         for (std::thread& worker : workers) {
130.             worker.join();
131.         }
132.     }
133. };
135. int main() {
136.     ThreadPool pool(2);
138.     // 高优先级任务
139.     pool.enqueue([&] {
140.         std::cout << "High priority task is running." << std::endl;
141.     }, 1);
143.     // 低优先级任务
144.     pool.enqueue([&] {
145.         std::cout << "Low priority task is running." << std::endl;
146.     }, 0);
148.     // 再添加一个高优先级任务
149.     pool.enqueue([&] {
150.         std::cout << "Another high priority task is running." << std::endl;
151.     }, 1);
153.     // 让主线程休眠一段时间，确保任务有足够的时间执行
154.     std::this_thread::sleep_for(std::chrono::seconds(2));
156.     return 0;
157. }
159. // 3.
160. #include <iostream>
161. #include <vector>
162. #include <queue>
163. #include <thread>
164. #include <mutex>
165. #include <condition_variable>
166. #include <functional>
167. #include <atomic>
168. #include <future>
169. #include <memory>
171. class ThreadPool {
172. private:
173.     std::vector<std::thread> workers;
174.     std::queue<std::function<void()>> tasks;
175.     std::mutex queue_mutex;
176.     std::condition_variable condition;
177.     std::atomic<bool> stop{ false };
179.     void worker() {
180.         while (true) {
181.             std::function<void()> task;
182.             {
183.                 std::unique_lock<std::mutex> lock(this->queue_mutex);
184.                 this->condition.wait(lock, [this] { return this->stop ||!this->tasks.empty(); });
185.                 if (this->stop && this->tasks.empty())
186.                     return;
187.                 task = std::move(this->tasks.front());
188.                 this->tasks.pop();
189.             }
190.             task();
191.         }
192.     }
194. public:
195.     ThreadPool(size_t numThreads) {
196.         for (size_t i = 0; i < numThreads; ++i) {
197.             workers.emplace_back([this] { this->worker(); });
198.         }
199.     }
201.     template<typename Func>
202.     auto enqueue(Func&& func) -> std::future<typename std::result_of<Func()>::type> {
203.         using return_type = typename std::result_of<Func()>::type;
205.         auto task = std::make_shared< std::packaged_task<return_type()> >(
206.             std::forward<Func>(func)
207.         );
208.         
209.         std::future<return_type> res = task->get_future();
210.         {
211.             std::unique_lock<std::mutex> lock(queue_mutex);
212.             if (stop)
213.                 throw std::runtime_error("enqueue on stopped ThreadPool");
214.             tasks.emplace([task]() { (*task)(); });
215.         }
216.         condition.notify_one();
217.         return res;
218.     }
220.     ~ThreadPool() {
221.         {
222.             std::unique_lock<std::mutex> lock(queue_mutex);
223.             stop = true;
224.         }
225.         condition.notify_all();
226.         for (std::thread& worker : workers) {
227.             worker.join();
228.         }
229.     }
230. };
233. int main() {
234.     // 创建一个包含 2 个线程的线程池
235.     ThreadPool pool(2);
237.     // 测试用例 1：返回一个整数
238.     auto fut1 = pool.enqueue([] { return 42; });
239.     std::cout << "Result 1: " << fut1.get() << std::endl;
241.     // 测试用例 2：返回一个字符串
242.     auto fut2 = pool.enqueue([] { return std::string("Hello, ThreadPool!"); });
243.     std::cout << "Result 2: " << fut2.get() << std::endl;
245.     // 测试用例 3：执行一个复杂一点的计算
246.     auto fut3 = pool.enqueue([] {
247.         int sum = 0;
248.         for (int i = 1; i <= 10; ++i) {
249.             sum += i;
250.         }
251.         return sum;
252.     });
253.     std::cout << "Result 3: " << fut3.get() << std::endl;
255.     return 0;
256. }
257. // 4.
258. #include <iostream>
259. #include <vector>
260. #include <queue>
261. #include <thread>
262. #include <mutex>
263. #include <condition_variable>
264. #include <functional>
265. #include <atomic>
266. #include <future>
267. #include <memory>
268. #include <chrono>
270. // 日志输出控制宏
271. #define ENABLE_LOGGING 1
272. #if ENABLE_LOGGING
273. #define LOG(msg) std::cout << msg << std::endl
274. #else
275. #define LOG(msg)
276. #endif
278. class ThreadPool {
279. private:
280.     struct Task {
281.         std::function<void()> func;
282.         int priority;
283.         bool operator<(const Task& other) const { return priority < other.priority; }
284.     };
286.     std::vector<std::thread> workers;
287.     std::priority_queue<Task> tasks;
288.     std::mutex queue_mutex;
289.     std::condition_variable condition;
290.     std::atomic<bool> stop{ false };
291.     std::atomic<size_t> current_size{ 0 };
292.     std::atomic<size_t> threads_to_exit{ 0 };
294.     void worker() {
295.         ++current_size;
296.         LOG("Thread started. Current size: " << current_size.load());
297.         while (true) {
298.             std::function<void()> task;
299.             {
300.                 std::unique_lock<std::mutex> lock(queue_mutex);
301.                 // 等待任务或退出信号
302.                 condition.wait(lock, [this] { return stop || threads_to_exit > 0 ||!tasks.empty(); });
304.                 // 如果需要退出且任务队列为空，则退出线程
305.                 if ((threads_to_exit > 0 && tasks.empty()) || stop) {
306.                     if (threads_to_exit > 0) {
307.                         --threads_to_exit;
308.                     }
309.                     --current_size;
310.                     LOG("Thread exited. Current size: " << current_size.load());
311.                     return;
312.                 }
314.                 task = std::move(tasks.top().func);
315.                 tasks.pop();
316.             }
317.             try {
318.                 task();
319.             } catch (const std::exception& e) {
320.                 LOG("Exception in task: " << e.what());
321.             }
322.         }
323.     }
325.     void add_threads(size_t num) {
326.         for (size_t i = 0; i < num; ++i) {
327.             workers.emplace_back([this] { worker(); });
328.         }
329.     }
331.     void remove_threads(size_t num) {
332.         threads_to_exit = num;
333.         condition.notify_all(); // 通知所有线程检查是否需要退出
334.         {
335.             std::unique_lock<std::mutex> lock(queue_mutex);
336.             // 等待线程退出
337.             while (current_size > workers.size() - num) {
338.                 condition.wait(lock);
339.             }
340.         }
341.         // 移除已退出的线程
342.         workers.erase(workers.begin() + (workers.size() - num), workers.end());
343.     }
345. public:
346.     explicit ThreadPool(size_t initial_size = 4) {
347.         resize(initial_size);
348.     }
350.     void resize(size_t new_size) {
351.         std::unique_lock<std::mutex> lock(queue_mutex);
352.         if (new_size > workers.size()) {
353.             add_threads(new_size - workers.size());
354.         } else if (new_size < workers.size()) {
355.             remove_threads(workers.size() - new_size);
356.         }
357.         LOG("Thread pool size resized to: " << workers.size());
358.     }
360.     template<typename Func>
361.     auto enqueue(Func&& func, int priority = 0) -> std::future<typename std::invoke_result<Func>::type> {
362.         using return_type = typename std::invoke_result<Func>::type;
364.         auto task = std::make_shared< std::packaged_task<return_type()> >(
365.             std::forward<Func>(func)
366.         );
368.         std::future<return_type> res = task->get_future();
369.         {
370.             std::unique_lock<std::mutex> lock(queue_mutex);
371.             if (stop)
372.                 throw std::runtime_error("Cannot enqueue task on stopped thread pool.");
373.             tasks.push({ [task]() { (*task)(); }, priority });
374.         }
375.         condition.notify_one();
376.         return res;
377.     }
379.     // 添加获取当前线程数量的方法
380.     size_t get_current_size() const {
381.         return workers.size();
382.     }
384.     // 检查任务队列是否为空，去掉 const 限定符
385.     bool is_task_queue_empty() {
386.         std::unique_lock<std::mutex> lock(queue_mutex);
387.         return tasks.empty();
388.     }
390.     ~ThreadPool() {
391.         {
392.             std::unique_lock<std::mutex> lock(queue_mutex);
393.             stop = true;
394.         }
395.         condition.notify_all();
396.         for (auto& worker : workers) {
397.             if (worker.joinable()) {
398.                 worker.join();
399.             }
400.         }
401.     }
402. };
404. // 任务函数增加中断检查
405. void long_running_task(std::atomic<bool>& interrupt_flag) {
406.     int count = 0;
407.     while (!interrupt_flag.load()) {
408.         // 模拟分段执行任务
409.         std::this_thread::sleep_for(std::chrono::milliseconds(200));
410.         LOG("Task is running, step: " << count++);
411.     }
412.     LOG("Task is interrupted.");
413. }
415. int main() {
416.     try {
417.         ThreadPool pool(2);
419.         LOG("Initial thread pool size: " << pool.get_current_size());
421.         // 增大线程池大小
422.         pool.resize(4);
423.         LOG("Thread pool size after increasing: " << pool.get_current_size());
425.         // 缩小线程池大小
426.         pool.resize(1);
427.         LOG("Thread pool size after decreasing: " << pool.get_current_size());
429.         std::atomic<bool> interrupt_flag(false);
430.         auto fut = pool.enqueue([&interrupt_flag] { long_running_task(interrupt_flag); });
432.         std::this_thread::sleep_for(std::chrono::seconds(1));
433.         interrupt_flag = true;
434.         fut.wait();
436.         // 检查任务队列是否为空
437.         if (pool.is_task_queue_empty()) {
438.             LOG("Task queue is empty.");
439.         } else {
440.             LOG("Task queue is not empty.");
441.         }
443.     } catch (const std::exception& e) {
444.         std::cerr << "Exception: " << e.what() << std::endl;
445.     }
447.     return 0;
448. }   
449. // 5.
450. #include <iostream>
451. #include <vector>
452. #include <queue>
453. #include <thread>
454. #include <mutex>
455. #include <condition_variable>
456. #include <functional>
457. #include <atomic>
458. #include <future>
459. #include <memory>
460. #include <chrono>
462. class ThreadPool {
463. private:
464.     struct Task {
465.         std::function<void()> func;
466.         int priority;
467.         bool operator<(const Task& other) const { return priority < other.priority; }
468.     };
470.     std::vector<std::thread> workers;
471.     std::priority_queue<Task> tasks;
472.     std::mutex queue_mutex;
473.     std::condition_variable condition;
474.     std::atomic<bool> stop{ false };
475.     std::atomic<size_t> current_size{ 0 };
476.     std::atomic<size_t> threads_to_exit{ 0 };
478.     void worker() {
479.         ++current_size;
480.         while (true) {
481.             std::function<void()> task;
482.             {
483.                 std::unique_lock<std::mutex> lock(queue_mutex);
484.                 condition.wait(lock, [this] { return stop || threads_to_exit > 0 ||!tasks.empty(); });
486.                 // 如果需要退出且任务队列为空，则退出线程
487.                 if ((threads_to_exit > 0 && tasks.empty()) || stop) {
488.                     if (threads_to_exit > 0) {
489.                         --threads_to_exit;
490.                     }
491.                     --current_size;
492.                     return;
493.                 }
495.                 task = std::move(tasks.top().func);
496.                 tasks.pop();
497.             }
498.             task();
499.         }
500.     }
502. public:
503.     explicit ThreadPool(size_t initial_size = 4) {
504.         resize(initial_size);
505.     }
507.     void resize(size_t new_size) {
508.         std::unique_lock<std::mutex> lock(queue_mutex);
509.         if (new_size > current_size) {
510.             for (size_t i = current_size; i < new_size; ++i) {
511.                 workers.emplace_back([this] { worker(); });
512.             }
513.         } else if (new_size < current_size) {
514.             threads_to_exit = current_size - new_size;
515.             condition.notify_all(); // 通知所有线程检查是否需要退出
516.             // 等待线程退出
517.             while (current_size > new_size) {
518.                 condition.wait(lock);
519.             }
520.             // 移除已退出的线程
521.             workers.erase(workers.begin() + new_size, workers.end());
522.         }
523.     }
525.     template<typename R>
526.     std::future<R> enqueue(std::function<R()> task, int priority = 0) {
527.         auto packaged = std::make_shared<std::packaged_task<R()>>(task);
528.         auto fut = packaged->get_future();
529.         {
530.             std::unique_lock<std::mutex> lock(queue_mutex);
531.             // 修改为使用 push 方法
532.             tasks.push({ [packaged] { (*packaged)(); }, priority });
533.         }
534.         condition.notify_one();
535.         return fut;
536.     }
538.     // 添加获取当前线程数量的方法
539.     size_t get_current_size() const {
540.         return current_size;
541.     }
543.     ~ThreadPool() {
544.         {
545.             std::unique_lock<std::mutex> lock(queue_mutex);
546.             stop = true;
547.         }
548.         condition.notify_all();
549.         for (auto& worker : workers) {
550.             if (worker.joinable()) {
551.                 worker.join();
552.             }
553.         }
554.     }
555. };
557. // 任务函数增加中断检查
558. void long_running_task(std::atomic<bool>& interrupt_flag) {
559.     int count = 0;
560.     while (!interrupt_flag.load()) {
561.         // 模拟分段执行任务
562.         std::this_thread::sleep_for(std::chrono::milliseconds(200));
563.         std::cout << "Task is running, step: " << count++ << std::endl;
564.     }
565.     std::cout << "Task is interrupted." << std::endl;
566. }
568. int main() {
569.     ThreadPool pool(2);
571.     // 定义中断标志
572.     std::atomic<bool> flag(false);
574.     // 显式转换 lambda 为 std::function
575.     auto task = std::function<void()>([&flag] { long_running_task(flag); });
576.     auto fut = pool.enqueue(task);
578.     // 主线程休眠一段时间，让任务执行一会儿
579.     std::this_thread::sleep_for(std::chrono::seconds(1));
581.     // 外部触发中断
582.     std::cout << "Triggering interruption..." << std::endl;
583.     flag.store(true);
585.     // 等待任务完成
586.     fut.wait();
588.     return 0;
589. }

复制代码

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。