在 QThreadPool 中执行槽

Question

我有一个应该在线程中运行的类，并且需要一个用于插槽的事件循环，目前我使用moveToThread() 很好地运行它，但我想使用QThreadPool，但我遇到了一个问题。

当使用QThreadPool 运行时，我的runnable 的run() 方法是从池线程中调用的（我使用QThread::currentThread() 进行了检查），但我的插槽没有在池线程中运行，所以我认为对象不会移动到池中的线程。

我认为这是因为我知道插槽是在接收器的线程中运行的，这正是我在使用 moveToThread() 方法和 QThread 时得到的（正确）行为。

如何让我的QRunnable（下例中的Foo）完全在池线程中运行？ 还是我做错了什么或理解错了？

以下 POC 演示了该问题：

foo.h

#ifndef FOO_H
#define FOO_H

#include <QObject>
#include <QRunnable>
#include <QEventLoop>

class Foo : public QObject, public QRunnable
{
    Q_OBJECT
public:
    explicit Foo(int data, QObject *parent = 0);
    void run();
signals:
    void startWorking();
public slots:
    void doWork();

private:
    QEventLoop eventLoop;
    int data;
};

#endif // FOO_H

foo.cpp

#include "foo.h"

#include <QThread>
#include <QDebug>

Foo::Foo(int d, QObject *parent) :
    QObject(parent), eventLoop(this), data(d)
{
}

void Foo::run()
{
    qDebug() << "run() in: " << QThread::currentThread();
    connect(this, SIGNAL(startWorking()), this, SLOT(doWork()));
    emit startWorking();
    eventLoop.exec();
}

void Foo::doWork()
{
    qDebug() << "doWork() in: " << QThread::currentThread();
}

ma​​in.cpp

#include <QCoreApplication>
#include <QThreadPool>

#include "foo.h"

int main(int argc, char *argv[])
{
    QCoreApplication a(argc, argv);

    Foo *foo = new Foo(42);

    QThreadPool::globalInstance()->start(foo);

    return a.exec();
}

但是请注意，在我的真实代码中，信号不会立即发出，因为它会在我收到网络上的一些数据之后发出。

PS：POC 也可以在here找到。

Answer 1

也许您可以将class Foo 中的逻辑拆分为两部分：
主机QRunnable 和QEventLoop 和一个工作线程QObject，您在调用QEventLoop::exec 方法之前在run() 的工作线程上创建它们。然后将所有信号转发给工作对象。 所以现在您的插槽将在池线程上调用。
但是，QThreadPool 是为执行大量短任务而设计的，而不会创建太多并发线程。一些任务已排队等待其他任务完成。如果这不是您的意图，您可能想回到旧的 QThread 并改用它。

Answer 2

您可以同时支持这两种模式，但需要外部协调。我的策略是从QRunnable::run 内部发出一个信号，传递当前线程。当您计划在线程池中使用它时，请在此信号上使用 Qt::BlockingQueuedConnection 并在那里执行您的 moveToThread。否则，将其移至QThread 并发出信号以开始正常工作。

TaskRunner.h

#pragma once

#include <QObject>
#include <QRunnable>
#include <QThread>

class TaskRunner : public QObject, public QRunnable
{
  Q_OBJECT
public:
  TaskRunner(int data, QObject* parent = nullptr);

  void run() override;

Q_SIGNALS:

  void start();
  void starting(QThread*);
  void stop();

private:
  int data;
};

TaskRunner.cpp

#include "TaskRunner.h"
#include <QEventLoop>
#include <stdexcept>


TaskRunner::TaskRunner(int data, QObject* parent)
: QObject(parent), data(data)
{
    // start should call run in the associated thread
    QObject::connect(this, &TaskRunner::start, this, &TaskRunner::run);
}

void TaskRunner::run()
{
    // in a thread pool, give a chance to move us to the current thread
    Q_EMIT starting(QThread::currentThread());

    if (thread() != QThread::currentThread())
      throw std::logic_error("Not associated with proper thread.");

    QEventLoop loop;
    QObject::connect(this, &TaskRunner::stop, &loop, &QEventLoop::quit);
    // other logic here perhaps
    loop.exec();
}

ma​​in.cpp

#include <QCoreApplication>
#include <QThreadPool>

#include "TaskRunner.h"

// comment to switch
#define USE_QTHREAD

int main(int argc, char *argv[])
{
    QCoreApplication a(argc, argv);

    auto runner = new TaskRunner(42);

#ifdef USE_QTHREAD
    // option 1: on a QThread
    auto thread = new QThread(&a);
    runner->moveToThread(thread);
    QObject::connect(thread, &QThread::finished, runner, &QObject::deleteLater);
    Q_EMIT runner->start();

    // stop condition not shown
#else
    // option 2: in a thread pool
    QObject::connect(
      runner, &TaskRunner::starting, 
      runner, &QObject::moveToThread,
      Qt::BlockingQueuedConnection);
    QThreadPool::globalInstance()->start(runner);

    // stop condition not shown
#endif

    return a.exec();
}

Answer 3

自从你的连接电话

connect(this, SIGNAL(startWorking()), this, SLOT(doWork()));

使用默认参数作为连接类型，它将是一个Qt::Autoconnection。 信号是从池化线程发出的，slot 仍然属于 foo，它与主线程有线程亲和性。自动连接会决定将 slot 放入主线程的事件队列中。

有两种方法可以解决此问题：

1.

connect(this, SIGNAL(startWorking()), this, SLOT(doWork()), Qt::DirectConnection);

并删除 eventloop.exec();

2.

在run方法中，在连接信号和槽之前将foo对象移动到当前线程。