[线程与线程池优化]: 升级多线程组件至C++20标准，增强功能与跨平台支持

RealChuan · RealChuan · commit 5af9bb55d263 · 2025-05-06T11:14:24.000+08:00
- **线程实现升级**: 将基础线程类从`std::thread`重构为`std::jthread`，利用C++20新特性简化线程管理逻辑，新增线程停止功能支持。
- **停止令牌集成**: 在所有线程相关操作中引入`std::stop_token`机制，实现线程任务的优雅中断，提升程序响应能力。
- **线程池优化**: 重构线程池内部任务队列，采用自定义模板队列替代原生队列，增强任务调度灵活性与性能表现。
- **跨平台适配调整**: 根据构建主机操作系统类型，调整线程组件的编译条件，移除非Windows系统对线程示例的构建支持（因`std::jthread`在Apple Clang下兼容性问题）。
- **单元测试扩展**: 新增线程与线程池相关测试用例共10项，涵盖线程启动/停止、任务添加、等待完成等核心功能验证。
- **资源管理改进**: 在线程池与任务队列中添加资源计数与完成状态通知机制，优化多线程环境下的资源监控与同步操作。
- **文档更新**: 修正README中线程模块描述，明确指出基于`std::jthread`实现，并标注Apple Clang平台的兼容性限制。
- **依赖基线更新**: 调整vcpkg依赖基线至最新版本，确保第三方组件与新线程API的兼容性。
diff --git a/.github/actions/install-dependencies/action.yml b/.github/actions/install-dependencies/action.yml
@@ -7,9 +7,6 @@ runs:
   steps:
     - name: Install Custom VCPKG
       uses: RealChuan/install-vcpkg@main
-      with:
-        repo: 'https://github.com/RealChuan/vcpkg.git'
-        branch: 'dev'
         
     - name: Update vcpkg manifest baseline
       shell: bash
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -2,7 +2,6 @@ cmake_minimum_required(VERSION 3.25.1)
 
 include(cmake/vcpkg.cmake)
 
-# 设定工程名
 project(
   Cpp-Examples
   VERSION 0.0.1
@@ -69,12 +68,14 @@ add_subdirectory(MonitorDir_EFSW)
 add_subdirectory(OpenSSL)
 add_subdirectory(SafeCallback)
 add_subdirectory(SpinMutex)
-add_subdirectory(Thread)
 
-if(CMAKE_HOST_LINUX)
+if(CMAKE_HOST_WIN32)
+  add_subdirectory(Thread)
+elseif(CMAKE_HOST_LINUX)
   add_subdirectory(Client)
   add_subdirectory(Icmp)
   add_subdirectory(Server)
+  add_subdirectory(Thread)
 endif()
 
 include(cmake/build_info.cmake)
diff --git a/README.md b/README.md
@@ -51,6 +51,6 @@
     2. [server_poll](/Server/server_poll.cc)——poll的例子；
     3. [server_select](/Server/server_select.cc)——select的例子；
 23. [SpinMutex](/SpinMutex)——使用std::atomic_flag实现的简单互斥锁和自旋锁；
-24. [Thread](/Thread/)——基于std::thread实现的线程类，包括线程池；
+24. [Thread](/Thread/)——基于std::jthread实现的线程类，包括线程池（`Apple Clang`不支持）；
     1. [Thread](/Thread/thread.hpp)——线程类；
     2. [ThreadPool](/Thread/threadpool.hpp)——线程池；
diff --git a/Thread/CMakeLists.txt b/Thread/CMakeLists.txt
@@ -3,8 +3,8 @@ target_link_libraries(thread_unittest PRIVATE GTest::gtest GTest::gtest_main
                                               GTest::gmock GTest::gmock_main)
 add_test(NAME thread_unittest COMMAND thread_unittest)
 
-add_executable(threadpool_unittest threadpool_unittest.cc threadpool.hpp
-                                   thread.hpp)
+add_executable(threadpool_unittest queue.hpp threadpool_unittest.cc
+                                   threadpool.hpp thread.hpp)
 target_link_libraries(
   threadpool_unittest PRIVATE GTest::gtest GTest::gtest_main GTest::gmock
                               GTest::gmock_main)
diff --git a/Thread/queue.hpp b/Thread/queue.hpp
@@ -0,0 +1,115 @@
+#pragma once
+
+#include <utils/object.hpp>
+
+#include <condition_variable>
+#include <functional>
+#include <mutex>
+#include <utility>
+#include <queue>
+
+template<typename T>
+    requires std::is_move_assignable_v<T>
+class Queue : noncopyable
+{
+    std::queue<T> m_queue;
+    bool m_stop = false;
+    mutable std::mutex m_mutex;
+    std::condition_variable m_condEmpty;
+    std::condition_variable m_condFull;
+    std::size_t m_maxSize = 0;
+
+public:
+    using PushSuccessCallback = std::function<void(void)>;
+
+    Queue() = default;
+    explicit Queue(std::size_t maxSize)
+        : m_maxSize(maxSize == 0 ? 1 : maxSize)
+    {}
+    ~Queue() { stop(); }
+
+    [[nodiscard]] auto push(T &&item, PushSuccessCallback callback = nullptr) -> bool
+    {
+        {
+            std::unique_lock lock(m_mutex);
+            m_condFull.wait(lock, [&]() { return !m_stop && (m_queue.size() < m_maxSize); });
+            if (m_stop) {
+                return false;
+            }
+            m_queue.push(std::move(item));
+            if (callback) {
+                callback();
+            }
+        }
+        m_condEmpty.notify_one();
+        return true;
+    }
+
+    [[nodiscard]] auto pop(T &item) -> bool
+    {
+        std::unique_lock lock(m_mutex);
+        m_condEmpty.wait(lock, [&]() { return !m_queue.empty() || m_stop; });
+        if (m_queue.empty()) {
+            return false;
+        }
+        item = std::move(m_queue.front());
+        m_queue.pop();
+        m_condFull.notify_one();
+        return true;
+    }
+
+    void setMaxSize(std::size_t maxSize)
+    {
+        std::scoped_lock guard(m_mutex);
+        m_maxSize = maxSize == 0 ? 1 : maxSize;
+        m_condFull.notify_all();
+    }
+
+    [[nodiscard]] auto getMaxSize() const -> std::size_t
+    {
+        std::scoped_lock guard(m_mutex);
+        return m_maxSize;
+    }
+
+    [[nodiscard]] auto size() const -> std::size_t
+    {
+        std::scoped_lock guard(m_mutex);
+        return m_queue.size();
+    }
+
+    [[nodiscard]] auto empty() const -> bool
+    {
+        std::scoped_lock guard(m_mutex);
+        return m_queue.empty();
+    }
+
+    void clear()
+    {
+        std::scoped_lock guard(m_mutex);
+        std::queue<T>().swap(m_queue);
+        m_condFull.notify_all();
+        m_condEmpty.notify_all();
+    }
+
+    [[nodiscard]] auto isStopped() const -> bool
+    {
+        std::scoped_lock guard(m_mutex);
+        return m_stop;
+    }
+
+    void start()
+    {
+        std::scoped_lock guard(m_mutex);
+        m_stop = false;
+    }
+
+    void stop()
+    {
+        {
+            std::scoped_lock guard(m_mutex);
+            m_stop = true;
+        }
+        m_condEmpty.notify_all();
+        m_condFull.notify_all();
+    }
+};
diff --git a/Thread/thread.hpp b/Thread/thread.hpp
@@ -7,72 +7,123 @@
 #include <chrono>
 #include <condition_variable>
 #include <functional>
+#include <iostream>
 #include <mutex>
 #include <thread>
 
 class Thread : noncopyable
 {
 public:
-    using Task = std::function<void()>;
+    using Task = std::function<void(std::stop_token)>;
 
     Thread() = default;
-    explicit Thread(Task task) { setTask(std::move(task)); }
+
+    explicit Thread(Task task)
+        : m_task(std::move(task))
+    {}
+
     ~Thread() { stop(); }
 
-    void setTask(Task task) { m_task = std::move(task); }
+    void setTask(Task task)
+    {
+        std::lock_guard<std::mutex> lock(m_mutex);
+        m_task = std::move(task);
+    }
 
-    void start()
+    bool start()
     {
+        if (m_jthread.joinable()) {
+            return false;
+        }
+
         m_running.store(true);
-        m_thread = std::thread([this]() {
-            m_condition.notify_one();
-            if (m_task) {
-                m_task();
+        m_jthread = std::jthread([this](std::stop_token token) {
+            {
+                std::unique_lock<std::mutex> lock(m_mutex);
+                if (m_task) {
+                    try {
+                        m_task(token);
+                    } catch (const std::exception &e) {
+                        std::cerr << "Thread exception: " << e.what() << std::endl;
+                    }
+                }
             }
+
+            m_running.store(false);
+            m_running.notify_all();
         });
+
+        return true;
     }
 
     void stop()
     {
-        if (m_thread.joinable()) {
-            m_thread.join();
+        if (m_jthread.joinable()) {
+            m_jthread.request_stop();
+            m_jthread.join();
+        }
+    }
+
+    void interrupt()
+    {
+        if (m_jthread.joinable()) {
+            m_jthread.request_stop();
         }
-        m_running.store(false);
     }
 
-    void waitForStarted()
+    void waitForFinished()
     {
-        std::unique_lock<std::mutex> lock(m_mutex);
-        m_condition.wait(lock, [this]() { return m_running.load(); });
+        while (m_running.load()) {
+            m_running.wait(true);
+        }
     }
 
-    [[nodiscard]] auto isRunning() const -> bool { return m_running; }
+    [[nodiscard]] auto isRunning() const -> bool { return m_running.load(); }
 
-    [[nodiscard]] auto getThreadId() const -> std::thread::id { return m_thread.get_id(); }
+    [[nodiscard]] auto getThreadId() const -> std::thread::id { return m_jthread.get_id(); }
 
     static void yield() { std::this_thread::yield(); }
 
+    static void sleep(unsigned int seconds)
+    {
+        std::this_thread::sleep_for(std::chrono::seconds(seconds));
+    }
+
+    static void msleep(unsigned int milliseconds)
+    {
+        std::this_thread::sleep_for(std::chrono::milliseconds(milliseconds));
+    }
+
+    static void usleep(unsigned int microseconds)
+    {
+        std::this_thread::sleep_for(std::chrono::microseconds(microseconds));
+    }
+
+    static void nsleep(unsigned int nanoseconds)
+    {
+        std::this_thread::sleep_for(std::chrono::nanoseconds(nanoseconds));
+    }
+
     static void sleepFor(std::chrono::milliseconds duration)
     {
         std::this_thread::sleep_for(duration);
     }
 
-    static void sleepUntil(std::chrono::system_clock::time_point timePoint)
+    static void sleepUntil(const std::chrono::system_clock::time_point &wake_time)
     {
-        std::this_thread::sleep_until(timePoint);
+        std::this_thread::sleep_until(wake_time);
     }
 
     static auto hardwareConcurrency() -> unsigned int
     {
-        auto availableCores = std::thread::hardware_concurrency(); // 如果不支持，返回0
+        auto availableCores = std::thread::hardware_concurrency();
         assert(availableCores > 0);
         return availableCores;
     }
 
 private:
-    std::thread m_thread;
+    std::jthread m_jthread;
     std::atomic_bool m_running{false};
     std::mutex m_mutex;
-    std::condition_variable m_condition;
     Task m_task;
 };
diff --git a/Thread/thread_unittest.cc b/Thread/thread_unittest.cc
@@ -2,29 +2,70 @@
 
 #include <gtest/gtest.h>
 
-TEST(Thread, start)
+void testTaskFunction(std::stop_token stopToken)
 {
-    Thread thread([]() { Thread::sleepFor(std::chrono::milliseconds(100)); });
+    for (int i = 0; i < 5; ++i) {
+        if (stopToken.stop_requested()) {
+            break;
+        }
+
+        std::cout << "Test task is running: " << i << std::endl;
+        Thread::msleep(200);
+    }
+}
+
+TEST(ThreadTest, StartAndStop)
+{
+    Thread thread;
+    thread.setTask(testTaskFunction);
     thread.start();
-    thread.waitForStarted();
     EXPECT_TRUE(thread.isRunning());
     thread.stop();
     EXPECT_FALSE(thread.isRunning());
 }
 
-TEST(Thread, setTask)
+TEST(ThreadTest, Interrupt)
 {
     Thread thread;
-    thread.setTask([]() { Thread::sleepFor(std::chrono::milliseconds(100)); });
+    thread.setTask(testTaskFunction);
     thread.start();
-    thread.waitForStarted();
     EXPECT_TRUE(thread.isRunning());
-    thread.stop();
+    thread.interrupt();
+    Thread::sleep(1);
+    EXPECT_FALSE(thread.isRunning());
+}
+
+TEST(ThreadTest, WaitForFinished)
+{
+    Thread thread;
+    thread.setTask(testTaskFunction);
+    thread.start();
+    thread.waitForFinished();
     EXPECT_FALSE(thread.isRunning());
 }
 
-auto main(int argc, char *argv[]) -> int
+TEST(ThreadTest, WaitForFinished2)
+{
+    Thread thread;
+    thread.waitForFinished();
+    EXPECT_FALSE(thread.isRunning());
+}
+
+TEST(ThreadTest, TaskExecution)
+{
+    Thread thread;
+    bool taskExecuted = false;
+    thread.setTask([&taskExecuted](std::stop_token) {
+        taskExecuted = true;
+        std::cout << "Task executed." << std::endl;
+    });
+    thread.start();
+    thread.stop();
+    EXPECT_TRUE(taskExecuted);
+}
+
+int main(int argc, char **argv)
 {
-    testing::InitGoogleTest(&argc, argv);
+    ::testing::InitGoogleTest(&argc, argv);
     return RUN_ALL_TESTS();
 }
diff --git a/Thread/threadpool.hpp b/Thread/threadpool.hpp
diff --git a/Thread/threadpool_unittest.cc b/Thread/threadpool_unittest.cc
diff --git a/vcpkg.json b/vcpkg.json
