[线程模块重构]: 基于 C++20 标准优化线程类及线程池实现

- 线程类优化：将底层实现从 `std::thread` 替换为 `std::jthread`，利用其内置的停止机制增强线程控制能力；调整线程启动逻辑，增加对停止请求的处理，提高线程安全性与可维护性
- 线程池改进：使用自定义 `Queue` 类替换原任务队列实现，提升任务管理效率与线程间通信能力；重构任务添加与执行流程，充分利用 `std::jthread` 特性优化线程调度与资源管理
- 依赖调整：更新 CMakeLists.txt 文件，修正线程模块在不同操作系统下的编译条件，确保代码在 Windows 平台的兼容性
- 文档更新：同步更新 README.md 中线程模块的描述，明确指出基于 `std::jthread` 实现且不支持 Apple Clang 编译器的特性说明
- 单元测试增强：完善线程与线程池的单元测试用例，覆盖启动、停止、中断、任务执行等关键场景，验证新实现的正确性与稳定性
- 代码结构优化：整理线程模块代码结构，规范化命名与注释，提升代码可读性与可扩展性，为后续功能开发奠定基础
- 性能提升：通过减少线程创建销毁开销、优化锁竞争机制，有效提高线程池在高并发场景下的执行效率与响应速度

Author: RealChuan

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)

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)——线程池；

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)

Thread/queue.hpp

@@ -0,0 +1,109 @@
+#pragma once
+
+#include <utils/object.hpp>
+
+#include <condition_variable>
+#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:
+    Queue() = default;
+    explicit Queue(std::size_t maxSize)
+        : m_maxSize(maxSize == 0 ? 1 : maxSize)
+    {}
+    ~Queue() { stop(); }
+
+    [[nodiscard]] auto push(T &&item) -> 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));
+        }
+        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();
+    }
+};

Thread/thread.hpp

@@ -7,48 +7,82 @@
 #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();
         }
+
         m_running.store(false);
     }
 
-    void waitForStarted()
+    void interrupt()
     {
-        std::unique_lock<std::mutex> lock(m_mutex);
-        m_condition.wait(lock, [this]() { return m_running.load(); });
+        if (m_jthread.joinable()) {
+            m_jthread.request_stop();
+        }
+    }
+
+    void waitForFinished()
+    {
+        while (m_running.load()) {
+            m_running.wait(true);
+        }
     }
 
-    [[nodiscard]] auto isRunning() const -> bool { return m_running; }
+    [[nodiscard]] bool isRunning() const { return m_running.load(); }
 
-    [[nodiscard]] auto getThreadId() const -> std::thread::id { return m_thread.get_id(); }
+    [[nodiscard]] std::thread::id getThreadId() const { return m_jthread.get_id(); }
 
     static void yield() { std::this_thread::yield(); }
 
@@ -64,15 +98,14 @@ class Thread : noncopyable
 
     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;
 };

Thread/thread_unittest.cc

@@ -2,29 +2,69 @@
 
 #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::sleepFor(std::chrono::milliseconds(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::sleepFor(std::chrono::seconds(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.sleepFor(std::chrono::seconds(1));
+    thread.stop();
+    EXPECT_TRUE(taskExecuted);
+}
+
+int main(int argc, char **argv)
 {
-    testing::InitGoogleTest(&argc, argv);
+    ::testing::InitGoogleTest(&argc, argv);
     return RUN_ALL_TESTS();
 }