含多函数的Singleton Worker Thread实现合理性及优化方案问询

问题描述 #

我想实现一个支持调用多函数的单例工作线程。之前见过的工作线程实现都只执行同一个函数，但我需要支持不同的函数调用。因为这些函数会共享部分资源，为每个函数单独创建工作线程不太方便。我现在的实现是加一个函数调用队列，把函数调用推入队列后触发条件变量。想请教这个实现是否可行，有没有更优的方式。

当前实现代码 #

#include <functional>
#include <queue>
#include <condition_variable>
#include <mutex>
#include <thread>

class Worker_Thread
{
    private:
        Worker_Thread() : th_{}, mtx_{}, q_{}
        {
            th_ = std::thread(&Worker_Thread::run, this);
        }

        std::thread th_;
        std::mutex mtx_;
        std::condition_variable cv_;
        std::queue<std::function<void()>> q_;

        void run();

        void func_a(int val) {};
        void func_b(float val) {};

    public:
        Worker_Thread(Worker_Thread const&) = delete;
        void operator=(Worker_Thread const&) = delete;

        static Worker_Thread& getInstance()
        {
            static Worker_Thread instance;
            return instance;
        }

        void do_a(int val);
        void do_b(float val);
};

void Worker_Thread::run()
{
    while (true) {
        std::unique_lock<std::mutex> lock(mtx_);
        cv_.wait(lock);

        std::function<void()> fnc = std::move(q_.front());
        fnc();
        q_.pop();
    }
}

void Worker_Thread::do_a(int val)
{
    {
        // Push a function onto the queue
        std::lock_guard<std::mutex> lock(mtx_);
        std::function<void()> fnc = std::bind(&Worker_Thread::func_a, this, val);
        q_.push(fnc);
    }
    // Notify the run thread to execute the queued function
    cv_.notify_one();
}

void Worker_Thread::do_b(float val)
{
    {
        // Push a function onto the queue
        std::lock_guard<std::mutex> lock(mtx_);
        std::function<void()> fnc = std::bind(&Worker_Thread::func_b, this, val);
        q_.push(fnc);
    }
    // Notify the run thread to execute the queued function
    cv_.notify_one();
}

int main()
{
    Worker_Thread& wth = Worker_Thread::getInstance();

    wth.do_a(1);
    wth.do_b(2.3);

    return 0;
}

可行性分析 #

你的实现是完全可行的，这是工业界常用的「单工作线程+任务队列」模型，完美适配你需要共享资源、多任务类型的场景：

• 队列实现了任务的异步提交与顺序执行，保证共享资源的线程安全（所有任务都在同一个线程执行，无需额外的资源锁）；
• 单例模式确保全局只有一个工作线程，避免了多线程资源竞争的复杂度。

现有实现的问题与改进点 #

1. 缺少优雅退出机制 #

当前run()是死循环，程序退出时工作线程会被强制终止，可能导致未完成任务丢失、资源泄漏。可以添加停止标志并在析构函数中处理：

• 新增std::atomic<bool> stop_{false};成员变量；
• 析构函数中设置stop_ = true并调用cv_.notify_one()；
• run()循环中检查stop_，处理完队列剩余任务再退出。

2. 未处理条件变量虚假唤醒 #

cv_.wait(lock)可能被虚假唤醒，导致线程尝试访问空队列引发未定义行为。需要给wait添加条件判断：

cv_.wait(lock, [this](){ return !q_.empty() || stop_; });

3. 任务异常未捕获 #

如果队列中的函数抛出未捕获的异常，会直接终止工作线程。需要在任务执行时添加异常捕获：

try {
    fnc();
} catch (const std::exception& e) {
    // 这里可以添加日志记录等异常处理逻辑
} catch (...) {
    // 捕获所有未知异常
}

4. 接口扩展性差 #

当前do_a、do_b是硬编码方法，新增函数需要修改类定义。可以提供通用的任务提交接口：

void submit(std::function<void()> task) {
    std::lock_guard<std::mutex> lock(mtx_);
    q_.push(std::move(task));
    cv_.notify_one();
}

外部可以直接提交任意可调用对象，无需新增方法：

wth.submit(std::bind(&Worker_Thread::func_a, &wth, 1));
// 或者用lambda更简洁
wth.submit([&wth](){ wth.func_b(2.3); });

更优实现方案（C++20+） #

如果你的项目支持C++20，可以用std::jthread替代std::thread，它自带线程终止管理，配合std::stop_token可以更优雅地处理线程退出：

#include <functional>
#include <queue>
#include <condition_variable>
#include <mutex>
#include <thread>
#include <atomic>

class Worker_Thread
{
private:
    Worker_Thread() : stop_{false} {
        th_ = std::jthread(&Worker_Thread::run, this);
    }

    std::jthread th_;
    std::mutex mtx_;
    std::condition_variable cv_;
    std::queue<std::function<void()>> q_;
    std::atomic<bool> stop_{false};

    void run(std::stop_token st) {
        while (!st.stop_requested()) {
            std::unique_lock<std::mutex> lock(mtx_);
            cv_.wait(lock, [this, &st](){ return !q_.empty() || st.stop_requested(); });

            if (st.stop_requested() && q_.empty()) break;

            auto fnc = std::move(q_.front());
            q_.pop();
            lock.unlock(); // 释放锁再执行任务，避免阻塞其他任务提交

            try {
                fnc();
            } catch (...) {
                // 异常处理
            }
        }
        // 处理剩余任务（可选）
        while (!q_.empty()) {
            auto fnc = std::move(q_.front());
            q_.pop();
            try {
                fnc();
            } catch (...) {}
        }
    }

    void func_a(int val) {};
    void func_b(float val) {};

public:
    Worker_Thread(const Worker_Thread&) = delete;
    void operator=(const Worker_Thread&) = delete;

    static Worker_Thread& getInstance() {
        static Worker_Thread instance;
        return instance;
    }

    void submit(std::function<void()> task) {
        std::lock_guard<std::mutex> lock(mtx_);
        q_.push(std::move(task));
        cv_.notify_one();
    }

    // 保留原有的do_a/do_b接口（可选）
    void do_a(int val) {
        submit(std::bind(&Worker_Thread::func_a, this, val));
    }

    void do_b(float val) {
        submit(std::bind(&Worker_Thread::func_b, this, val));
    }
};

内容的提问来源于stack exchange，提问作者SillyGoose