修改完善更新线程池实现的内容与解释 #12

Author: Mq-b

md/详细分析/04线程池.md

@@ -119,7 +119,7 @@ int main() {
 
 - 线程池中的线程会从任务队列中取出任务并执行，任务执行完毕后，线程继续取下一个任务或者休眠。
 
-- 调用 join 方法等待所有任务执行完毕并关闭线程池（`thread_pool` 析构同样会调用它，非必要）。
+- 调用 join 方法等待所有任务执行完毕并关闭线程池。
 
 如果我们不自己指明线程池的线程数量，那么 Asio 会根据函数 [`default_thread_pool_size`](https://github.com/boostorg/asio/blob/44238d033e1503c694782925d647811380a067c2/include/boost/asio/impl/thread_pool.ipp#L53-L58) 计算并返回一个**线程池的默认线程数量**。它根据系统的硬件并发能力来决定使用的线程数，通常是硬件并发能力的两倍。
 
@@ -151,12 +151,26 @@ thread_pool::~thread_pool()
 }
 ```
 
-- `stop` ：修改内部的标志位存在使得线程池能够识别何时需要停止接收新的任务，然后唤醒所有线程。
+- `stop` ：修改内部的标志位存在使得线程池能够识别何时需要停止接收新的任务，以及关闭还没开始执行的任务，然后唤醒所有线程。
 - `join()` ：等待所有线程完成它们的工作，确保所有线程都已终止。
 - `shutdown()` ：进行最终的清理，释放资源，确保线程池的完全清理和资源的正确释放
 
 > 此处可阅读部分源码，帮助理解与记忆
 
+析构函数先调用了 `stop()` ，然后再进行 `join()` 。那如果我们没有提前显式调用 `join()` 成员函数，**可能导致一些任务没有执行，析构函数并不会等待所有任务执行完毕**：
+
+```cpp
+boost::asio::thread_pool pool{ 4 }; 
+
+for (int i = 0; i < 10; ++i) {
+    boost::asio::post(pool, [i]() { print_task(i); });
+}
+```
+
+> [运行](https://godbolt.org/z/haPqKb1h7)测试。
+
+因为析构函数并不是堵塞执行完所有任务，而是先**停止**，再 `join()` 以及 `shutdown()`。
+
 `Boost.Asio` 提供的线程池使用十分简单，接口高度封装，几乎无需关心底层具体实现，易于使用。
 
 我们的操作几乎只需创建线程池对象、将任务加入线程池、在需要时调用 `join()`。
@@ -257,7 +271,7 @@ threadPool->start([=]{
 
 ## 实现线程池
 
-实现一个普通的能够满足日常开发需求的线程池实际上非常简单，也只需要一百多行代码。
+实现一个普通的能够满足日常开发需求的线程池实际上非常简单，只需要不到一百行代码。
 
 > - “*普通的能够满足日常开发需*求的”
 >
@@ -315,14 +329,30 @@ public:
         : stop_{ false }, num_threads_{ num_thread } {
         start();
     }
-    ~ThreadPool(){
+
+    ~ThreadPool() {
         stop();
+        join();
+    }
+
+    void stop() {
+        stop_.store(true);
+        cv_.notify_all();
+    }
+
+    void join() {
+        for (auto& thread : pool_) {
+            if (thread.joinable()) {
+                thread.join();
+            }
+        }
+        pool_.clear();
     }
 
     template<typename F, typename... Args>
-    std::future<std::invoke_result_t<std::decay_t<F>, std::decay_t<Args>...>> submit(F&& f, Args&&...args){
+    std::future<std::invoke_result_t<std::decay_t<F>, std::decay_t<Args>...>> submit(F&& f, Args&&...args) {
         using RetType = std::invoke_result_t<std::decay_t<F>, std::decay_t<Args>...>;
-        if (stop_.load()){
+        if (stop_.load()) {
             throw std::runtime_error("ThreadPool is stopped");
         }
 
@@ -338,41 +368,103 @@ public:
         return ret;
     }
 
-    void stop(){
-        stop_.store(true);
-        cv_.notify_all();
-        for (auto& thread : pool_){
-            if (thread.joinable()) {
-                thread.join();
-            }
-        }
-    }
-
-    void start(){
-        for (std::size_t i = 0; i < num_threads_; ++i){
+    void start() {
+        for (std::size_t i = 0; i < num_threads_; ++i) {
             pool_.emplace_back([this] {
                 while (!stop_) {
                     Task task;
                     {
                         std::unique_lock<std::mutex> lc{ mutex_ };
-                        cv_.wait(lc, [this] {return stop_ || !tasks_.empty(); });
                         if (tasks_.empty())
                             return;
+                        cv_.wait(lc, [this] {return stop_ || !tasks_.empty(); });
                         task = std::move(tasks_.front());
                         tasks_.pop();
                     }
                     task();
                 }
-            });
+                });
         }
     }
 
 private:
     std::mutex                mutex_;
     std::condition_variable   cv_;
-    std::atomic<bool>         stop_ ;
+    std::atomic<bool>         stop_;
     std::atomic<std::size_t>  num_threads_;
     std::queue<Task>          tasks_;
     std::vector<std::thread>  pool_;
 };
 ```
+
+**标头依赖**：
+
+```cpp
+#include <iostream>
+#include <thread>
+#include <mutex>
+#include <condition_variable>
+#include <future>
+#include <atomic>
+#include <queue>
+#include <vector>
+#include <syncstream>
+#include <functional>
+```
+
+**测试 demo**：
+
+```cpp
+void print_task(int n) {
+    std::osyncstream{ std::cout } << "Task " << n << " is running." << std::endl;
+}
+void print_task2(int n) {
+    std::osyncstream{ std::cout } << "🐢🐢🐢 " << n << " 🐉🐉🐉" << std::endl;
+}
+
+int main() {
+    ThreadPool pool{ 4 }; // 创建一个有 4 个线程的线程池
+
+    for (int i = 0; i < 10; ++i) {
+        pool.submit(print_task, i);
+    }
+    pool.join(); // 确保执行完所有任务
+
+    std::puts("---------------------");
+
+    pool.start(); // 重新启动线程池
+
+    for (int i = 0; i < 10; ++i) {
+        pool.submit(print_task2, i);
+    }
+    pool.join(); // 确保执行完所有任务
+} // 析构自动 stop() join()
+```
+
+**可能的[运行结果](https://godbolt.org/z/rM6vz1s9v)**：
+
+```shell
+Task 1 is running.
+Task 2 is running.
+Task 0 is running.
+Task 3 is running.
+Task 4 is running.
+Task 5 is running.
+Task 6 is running.
+Task 7 is running.
+Task 8 is running.
+Task 9 is running.
+---------------------
+🐢🐢🐢 0 🐉🐉🐉
+🐢🐢🐢 2 🐉🐉🐉
+🐢🐢🐢 1 🐉🐉🐉
+🐢🐢🐢 5 🐉🐉🐉
+🐢🐢🐢 7 🐉🐉🐉
+🐢🐢🐢 3 🐉🐉🐉
+🐢🐢🐢 4 🐉🐉🐉
+🐢🐢🐢 6 🐉🐉🐉
+🐢🐢🐢 8 🐉🐉🐉
+🐢🐢🐢 9 🐉🐉🐉
+```
+
+如果不自己显式调用 `join()` ，而是等待线程池析构调用，那么效果如同 `asio::thread_pool`，会先进行 `stop`，导致一些任务不执行。