初步完成基础线程池的实现 Mq-b#12

Author: Mq-b

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

@@ -267,34 +267,112 @@ threadPool->start([=]{
 
 ```cpp
 class ThreadPool {
-    std::mutex                mutex;
-    std::condition_variable   cv;
-    std::atomic<bool>         stop;
-    std::atomic<std::size_t>  thread_num;
-    std::queue<Task>          tasks;
-    std::vector<std::thread>  pool;
+    std::mutex                mutex_;
+    std::condition_variable   cv_;
+    std::atomic<bool>         stop_;
+    std::atomic<std::size_t>  num_threads_;
+    std::queue<Task>          tasks_;
+    std::vector<std::thread>  pool_;
 };
 ```
 
-1. **`std::mutex mutex`**
-
+1. **`std::mutex mutex_`**
    - 用于保护共享资源（如任务队列）在多线程环境中的访问，避免数据竞争。
 
-2. **`std::condition_variable cv`**
-
+2. **`std::condition_variable cv_`**
    - 用于线程间的同步，允许线程等待特定条件（如新任务加入队列）并在条件满足时唤醒线程。
 
-3. **`std::atomic<bool> stop`**
+3. **`std::atomic<bool> stop_`**
    - 用于指示线程池是否停止接收新任务，并安全地通知所有工作线程退出。
 
-4. **`std::atomic<std::size_t> thread_num`**
-
+4. **`std::atomic<std::size_t> num_threads_`**
    - 表示线程池中的线程数量。
 
-5. **`std::queue<Task> tasks`**
-
+5. **`std::queue<Task> tasks_`**
    - 任务队列，存储等待执行的任务，任务按提交顺序执行。
 
-6. **`std::vector<std::thread> pool`**
+6. **`std::vector<std::thread> pool_`**
 
    - 线程容器，存储管理线程对象，每个线程从任务队列中获取任务并执行。
+
+那么再直接提供构造函数以及添加任务到线程池的接口，然后内部再进行一点小小的操作，也就完成了：
+
+```cpp
+inline std::size_t default_thread_pool_size()noexcept {
+    std::size_t num_threads = std::thread::hardware_concurrency() * 2;
+    num_threads = num_threads == 0 ? 2 : num_threads;
+    return num_threads;
+}
+
+class ThreadPool {
+public:
+    using Task = std::packaged_task<void()>;
+
+    ThreadPool(const ThreadPool&) = delete;
+    ThreadPool& operator=(const ThreadPool&) = delete;
+
+    ThreadPool(std::size_t num_thread = default_thread_pool_size())
+        : stop_{ false }, num_threads_{ num_thread } {
+        start();
+    }
+    ~ThreadPool(){
+        stop();
+    }
+
+    template<typename F, typename... 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()){
+            throw std::runtime_error("ThreadPool is stopped");
+        }
+
+        auto task = std::make_shared<std::packaged_task<RetType()>>(
+            std::bind(std::forward<F>(f), std::forward<Args>(args)...));
+        std::future<RetType> ret = task->get_future();
+
+        {
+            std::lock_guard<std::mutex> lc{ mutex_ };
+            tasks_.emplace([task] {(*task)(); });
+        }
+        cv_.notify_one();
+        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){
+            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;
+                        task = std::move(tasks_.front());
+                        tasks_.pop();
+                    }
+                    task();
+                }
+            });
+        }
+    }
+
+private:
+    std::mutex                mutex_;
+    std::condition_variable   cv_;
+    std::atomic<bool>         stop_ ;
+    std::atomic<std::size_t>  num_threads_;
+    std::queue<Task>          tasks_;
+    std::vector<std::thread>  pool_;
+};
+```