Qt异步TaskRunner回调切换至主线程执行的实现方案问询

核心思路 #

• 利用Qt的QMetaObject::invokeMethod结合Qt::QueuedConnection，将回调函数的执行切换到主线程的事件循环中。
• 借助QCoreApplication::instance()作为接收对象，它始终运行在主线程，确保回调在正确的线程执行。

修改后的TaskRunner完整代码 #

#include <QThreadPool>
#include <QScopedPointer>
#include <QMutex>
#include <QtConcurrent>
#include <QCoreApplication>
#include <variant>
#include <functional>
#include <tuple>

// 辅助trait：判断T是否是指定模板类的实例
template<typename T, typename U>
struct is_instance : std::false_type {};

// 特化：匹配std::variant
template<typename... Ts>
struct is_instance<std::variant<Ts...>, std::variant> : std::true_type {};

// 特化：匹配std::optional
template<typename T>
struct is_instance<std::optional<T>, std::optional> : std::true_type {};

class TaskRunner
{
public:
    TaskRunner() :
        mPool(new QThreadPool())
    {
        mPool->setMaxThreadCount(1);
        mPool->setExpiryTimeout(-1);
    }

public:
    // 处理返回std::variant的异步任务
    template <typename Func, typename OnComplete, typename OnError, typename... Args>
    std::enable_if_t<is_instance<std::invoke_result_t<Func, Args...>, std::variant>::value>
    async(Func func, OnComplete onComplete, OnError onError, Args... args)
    {
        QMutexLocker lock(&mutex);
        // 打包参数并捕获，避免拷贝冗余
        (void)QtConcurrent::run(mPool.get(), [func, onComplete, onError, args = std::make_tuple(args...)]() mutable {
            const auto result = std::apply(std::move(func), std::move(args));
            
            if (result.index() == 0) {
                auto completeRes = std::get<0>(result);
                // 提交onComplete到主线程执行
                QMetaObject::invokeMethod(QCoreApplication::instance(), [onComplete, res = std::move(completeRes)]() mutable {
                    onComplete(std::move(res));
                }, Qt::QueuedConnection);
            } else {
                auto errorRes = std::get<1>(result);
                // 提交onError到主线程执行
                QMetaObject::invokeMethod(QCoreApplication::instance(), [onError, err = std::move(errorRes)]() mutable {
                    onError(std::move(err));
                }, Qt::QueuedConnection);
            }
        });
    }

    // 处理返回std::optional的异步任务
    template <typename Func, typename OnComplete, typename OnError, typename... Args>
    std::enable_if_t<is_instance<std::invoke_result_t<Func, Args...>, std::optional>::value>
    async(Func func, OnComplete onComplete, OnError onError, Args... args)
    {
        QMutexLocker lock(&mutex);
        (void)QtConcurrent::run(mPool.get(), [func, onComplete, onError, args = std::make_tuple(args...)]() mutable {
            const auto result = std::apply(std::move(func), std::move(args));
            
            if (result.has_value()) {
                auto completeRes = *result;
                QMetaObject::invokeMethod(QCoreApplication::instance(), [onComplete, res = std::move(completeRes)]() mutable {
                    onComplete(std::move(res));
                }, Qt::QueuedConnection);
            } else {
                QMetaObject::invokeMethod(QCoreApplication::instance(), [onError]() {
                    onError();
                }, Qt::QueuedConnection);
            }
        });
    }

private:
    QScopedPointer<QThreadPool> mPool;
    QMutex mutex;
};

使用示例 #

#include <QApplication>
#include <iostream>

int main(int argc, char *argv[])
{
    QApplication app(argc, argv); // 必须初始化Qt应用对象

    TaskRunner runner;
    // 正常任务
    runner.async([](int a, int b) -> std::variant<int, std::string> {
        if(b != 0)
            return a / b;
        else
            return std::string("division by zero");
    },
    [](int result){ 
        std::cout << "计算结果: " << result << '\n'; 
        // 此处可安全操作UI组件，因为在主线程
    },
    [](const std::string& error){ 
        std::cout << "错误信息: " << error << '\n'; 
    },
    10, 2);

    // 触发错误的任务
    runner.async([](int a, int b) -> std::variant<int, std::string> {
        return std::string("division by zero");
    },
    [](int result){ 
        std::cout << "计算结果: " << result << '\n'; 
    },
    [](const std::string& error){ 
        std::cout << "错误信息: " << error << '\n'; 
        app.quit(); // 在主线程退出应用
    },
    10, 0);

    return app.exec();
}

关键说明 #

• 线程切换原理：Qt::QueuedConnection会将方法调用放入接收对象所在线程的事件队列，QCoreApplication::instance()运行在主线程，因此回调会在主线程执行。
• 参数优化：使用std::make_tuple和std::apply实现参数的完美转发，减少不必要的拷贝开销。
• 生命周期管理：回调捕获的变量采用值传递，避免悬空引用问题；如果需要捕获大对象，可配合智能指针使用。
• 前置条件：必须确保程序初始化了QCoreApplication或QApplication，否则QCoreApplication::instance()返回空指针，导致回调无法执行。

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