如何在Qt中使用排队连接时压缩插槽调用？

Question

读了一些文章后,像这样关于Qt信号插槽通信还是有关于排队的连接问题.

如果我有一些线程一直向对方发送信号,并且假设一个thread_slow在它的事件循环中运行一个慢速方法而另一个thread_fast运行一个快速的方法发送多个信号,而另一个线程仍在运行它的慢速方法...... ..当thread_slow返回到事件循环的慢速方法时,它会处理之前发送的所有信号thread_fast还是最后一个发送的所有信号(所有信号都是相同的类型)？

如果它将处理所有信号,是否有办法使thread_slow 唯一的过程成为最后一个？(考虑到多线程应用程序中的"最后一个"可能是模糊的,让我们考虑线程要求最后一个信号之前的最后一个信号,为了简单起见,所以在线程查找最后一个时发送的新信号可能会丢失).

(我问这个是因为我有多个线程从多个线程接收数据,我不希望它们处理旧数据,只是最后一个发送的数据)

我已经进行了一些测试,看来Qt将处理所有信号.我做了一个线程:

while(true)
{
    QThread::msleep(500);
    emit testQueue(test);
    test++;
}

另一个插槽将执行:

void test::testQueue(int test)
{
    test.store(private_test.load() + test);
    emit testText(QString("Test Queue: ") + QString::number(private_test.load()));
}

并且线程将运行:

while(true)
{
    QThread::msleep(3000);
    QCoreApplication::processEvents();
    private_test.store(private_test.load() + 1000);
}

我每隔500毫秒从一个线程向另一个线程发送一个信号,另一个线程休眠3000毫秒(3秒)然后唤醒并将内部变量递增100.每次执行一个插槽时它会发出一个文本收到的值+内部变量.我得到的结果是,每次 QCoreApplication::processEvents();调用,所有信号都被执行....(我编辑了这部分是因为我在之前的代码中发现了一个错误)

Answer 1

QCoreApplication QMetaCallEvent压缩

每个排队的插槽调用都会在QMetaCallEvent向目标对象发布a时结束.该事件包含发送方对象,信号ID,插槽索引和打包的调用参数.在Qt 5上,信号id通常不等于返回的值QMetaObject::signalIndex():它是一个计算的索引,好像该对象只有信号方法而没有其他方法.

目标是压缩这样的调用,以便在给定元组(发送者对象,发送者信号,接收者对象,接收者时隙)的事件队列中仅存在一个唯一的调用.

这是唯一理智的方法,无需更改源或目标对象,同时保持最小的开销.我的其他答案中的事件循环递归方法每个事件都有严重的堆栈开销,当为64位指针体系结构构建Qt时,大约为1kbyte.

当新事件发布到已经发布了一个或多个事件的对象时,可以访问事件队列.在这种情况下,QCoreApplication::postEvent电话QCoreApplication::compressEvent.compressEvent第一个事件发布到对象时不会调用.在重新实现此方法时,QMetaCallEvent可以检查发布到目标对象的内容以查询对您的插槽的调用,并且必须删除过时的副本.私人Qt的标题必须包含获取的定义QMetaCallEvent,QPostEvent和QPostEventList.

优点:发送者和接收者对象都不必知道任何事情.信号和插槽按原样工作,包括Qt 5中的方法指针调用.Qt本身使用这种压缩事件的方式.

缺点:需要包含私有Qt标头和强制清除QEvent::posted标志.

当一个零持续时间计时器被触发时QEvent::posted,要删除的事件可以在一个单独的列表中排队,并在compressEvent调用之外删除,而不是黑客攻击该标志.这具有额外的事件列表的开销,并且每个事件删除迭代通过发布的事件列表.

其他方法

以其他方式做这件事的关键是不要使用Qt的内部.

L1第一个限制是无法访问私人定义的内容QMetaCallEvent.它可以处理如下:

1. 具有与目标相同签名的信号和槽的代理对象可以连接在源对象和目标对象之间.

2. QMetaCallEvent在代理对象上运行允许提取调用类型,被调用的槽ID和参数.

3. 代替信号槽连接,可以将事件明确地发布到目标对象.目标对象或事件过滤器必须从事件的数据中明确地重新合成槽调用.

4. compressedConnect可以使用自定义实现来代替QObject::connect.这充分暴露了信号和插槽的细节.代理对象可用于queued_activate在发送方对象的一侧执行压缩友好的等效项.

L2第二个限制是无法完全重新实现QCoreApplication::compressEvent,因为事件列表是私有定义的.我们仍然可以访问被压缩的事件,我们仍然可以决定是否删除它,但是没有办法迭代事件列表.从而:

1. 事件队列可以被隐式通过递归调用访问sendPostedEvents从内notify(从由此也eventFilter(),event()或者从槽).这不会导致死锁,因为QCoreApplication::sendPostedEvents在事件通过时传递时不能(也不会)保存事件循环互斥锁sendEvent.事件可以按如下方式过滤:

   • 全球重新实现QCoreApplication::notify,
   • 全球通过注册QInternal::EventNotifyCallback,
   • 在本地通过附加事件过滤器到对象,
   • 通过QObject::event()在目标类中重新实现来显式本地化.

   重复事件仍会发布到事件队列中.notify来自内部的递归调用sendPostedEvents消耗了相当多的堆栈空间(64位指针体系结构上的预算为1kb).

2. QCoreApplication::removePostedEvents在将新事件发布到对象之前,可以通过调用删除已存在的事件.不幸的是,QCoreApplication::compressEvent由于事件队列互斥锁已经被占用,因此执行此操作会导致死锁.

   包含指向receiver对象的指针的自定义事件类可以自动调用removePostedEvents构造函数.

3. QEvent::Exit可以重新使用现有的压缩事件,例如.

   这些事件的集合是一个实现细节,可能会改变.除了接收器QObject指针之外,Qt不区分那些事件.实现需要每个(事件类型,接收器对象)元组的代理QObject的开销.

履行

下面的代码适用于Qt 4和Qt 5.对于后者,请确保添加QT += core-private到qmake项目文件中,以便包含私有Qt标头.

不使用Qt内部头文件的实现在其他答案中给出:

• 使用L1.1和L2.1.
• 使用L1.2和L2.1.

选择了两个事件删除代码路径if (true).启用的代码路径通常会保留最近的事件并且最有意义.或者,您可能希望保留最早的事件 - 这是禁用的代码路径所执行的操作.

#include <QApplication>
#include <QMap>
#include <QSet>
#include <QMetaMethod>
#include <QMetaObject>
#include <private/qcoreapplication_p.h>
#include <private/qthread_p.h>
#include <private/qobject_p.h>

#include <QWidget>
#include <QPushButton>
#include <QPlainTextEdit>
#include <QSpinBox>
#include <QFormLayout>

// Works on both Qt 4 and Qt 5.

//
// Common Code

/*! Keeps a list of singal indices for one or more meatobject classes.
 * The indices are signal indices as given by QMetaCallEvent.signalId.
 * On Qt 5, those do *not* match QMetaObject::methodIndex since they
 * exclude non-signal methods. */
class SignalList {
    Q_DISABLE_COPY(SignalList)
    typedef QMap<const QMetaObject *, QSet<int> > T;
    T m_data;
    /*! Returns a signal index that is can be compared to QMetaCallEvent.signalId. */
    static int signalIndex(const QMetaMethod & method) {
        Q_ASSERT(method.methodType() == QMetaMethod::Signal);
#if QT_VERSION >= QT_VERSION_CHECK(5,0,0)
        int index = -1;
        const QMetaObject * mobj = method.enclosingMetaObject();
        for (int i = 0; i <= method.methodIndex(); ++i) {
            if (mobj->method(i).methodType() != QMetaMethod::Signal) continue;
            ++ index;
        }
        return index;
#else
        return method.methodIndex();
#endif
    }
public:
    SignalList() {}
    void add(const QMetaMethod & method) {
        m_data[method.enclosingMetaObject()].insert(signalIndex(method));
    }
    void remove(const QMetaMethod & method) {
        T::iterator it = m_data.find(method.enclosingMetaObject());
        if (it != m_data.end()) {
            it->remove(signalIndex(method));
            if (it->empty()) m_data.erase(it);
        }
    }
    bool contains(const QMetaObject * metaObject, int signalId) {
        T::const_iterator it = m_data.find(metaObject);
        return it != m_data.end() && it.value().contains(signalId);
    }
};

//
// Implementation Using Event Compression With Access to Private Qt Headers

struct EventHelper : private QEvent {
    static void clearPostedFlag(QEvent * ev) {
        (&static_cast<EventHelper*>(ev)->t)[1] &= ~0x8001; // Hack to clear QEvent::posted
    }
};

template <class Base> class CompressorApplication : public Base {
    SignalList m_compressedSignals;
public:
    CompressorApplication(int & argc, char ** argv) : Base(argc, argv) {}
    void addCompressedSignal(const QMetaMethod & method) { m_compressedSignals.add(method); }
    void removeCompressedSignal(const QMetaMethod & method) { m_compressedSignals.remove(method); }
protected:
    bool compressEvent(QEvent *event, QObject *receiver, QPostEventList *postedEvents) {
        if (event->type() != QEvent::MetaCall)
            return Base::compressEvent(event, receiver, postedEvents);

        QMetaCallEvent *mce = static_cast<QMetaCallEvent*>(event);
        if (! m_compressedSignals.contains(mce->sender()->metaObject(), mce->signalId())) return false;
        for (QPostEventList::iterator it = postedEvents->begin(); it != postedEvents->end(); ++it) {
            QPostEvent &cur = *it;
            if (cur.receiver != receiver || cur.event == 0 || cur.event->type() != event->type())
                continue;
            QMetaCallEvent *cur_mce = static_cast<QMetaCallEvent*>(cur.event);
            if (cur_mce->sender() != mce->sender() || cur_mce->signalId() != mce->signalId() ||
                    cur_mce->id() != mce->id())
                continue;
            if (true) {
              /* Keep The Newest Call */              
              // We can't merely qSwap the existing posted event with the new one, since QEvent
              // keeps track of whether it has been posted. Deletion of a formerly posted event
              // takes the posted event list mutex and does a useless search of the posted event
              // list upon deletion. We thus clear the QEvent::posted flag before deletion.
              EventHelper::clearPostedFlag(cur.event);
              delete cur.event;
              cur.event = event;
            } else {
              /* Keep the Oldest Call */
              delete event;
            }
            return true;
        }
        return false;
    }
};

//
// Demo GUI

class Signaller : public QObject {
    Q_OBJECT
public:
    Q_SIGNAL void emptySignal();
    Q_SIGNAL void dataSignal(int);
};

class Widget : public QWidget {
    Q_OBJECT
    QPlainTextEdit * m_edit;
    QSpinBox * m_count;
    Signaller m_signaller;
    Q_SLOT void emptySlot() {
        m_edit->appendPlainText("emptySlot invoked");
    }
    Q_SLOT void dataSlot(int n) {
        m_edit->appendPlainText(QString("dataSlot(%1) invoked").arg(n));
    }
    Q_SLOT void sendSignals() {
        m_edit->appendPlainText(QString("\nEmitting %1 signals").arg(m_count->value()));
        for (int i = 0; i < m_count->value(); ++ i) {
            emit m_signaller.emptySignal();
            emit m_signaller.dataSignal(i + 1);
        }
    }
public:
    Widget(QWidget * parent = 0) : QWidget(parent),
        m_edit(new QPlainTextEdit), m_count(new QSpinBox)
    {
        QFormLayout * l = new QFormLayout(this);
        QPushButton * invoke = new QPushButton("Invoke");
        m_edit->setReadOnly(true);
        m_count->setRange(1, 1000);
        l->addRow("Number of slot invocations", m_count);
        l->addRow(invoke);
        l->addRow(m_edit);
#if QT_VERSION >= QT_VERSION_CHECK(5,0,0)
        connect(invoke, &QPushButton::clicked, this, &Widget::sendSignals);
        connect(&m_signaller, &Signaller::emptySignal, this, &Widget::emptySlot, Qt::QueuedConnection);
        connect(&m_signaller, &Signaller::dataSignal, this, &Widget::dataSlot, Qt::QueuedConnection);
#else
        connect(invoke, SIGNAL(clicked()), SLOT(sendSignals()));
        connect(&m_signaller, SIGNAL(emptySignal()), SLOT(emptySlot()), Qt::QueuedConnection);
        connect(&m_signaller, SIGNAL(dataSignal(int)), SLOT(dataSlot(int)), Qt::QueuedConnection);
#endif
    }
};

int main(int argc, char *argv[])
{
    CompressorApplication<QApplication> a(argc, argv);
#if QT_VERSION >= QT_VERSION_CHECK(5,0,0)
    a.addCompressedSignal(QMetaMethod::fromSignal(&Signaller::emptySignal));
    a.addCompressedSignal(QMetaMethod::fromSignal(&Signaller::dataSignal));
#else
    a.addCompressedSignal(Signaller::staticMetaObject.method(Signaller::staticMetaObject.indexOfSignal("emptySignal()")));
    a.addCompressedSignal(Signaller::staticMetaObject.method(Signaller::staticMetaObject.indexOfSignal("dataSignal(int)")));
#endif
    Widget w;
    w.show();
    return a.exec();
}

#include "main.moc"

Answer 2

我试图将我的评论纳入答案。我同意您的意见，即文档缺少此信息，或者至少对于我来说还是不清楚，或者对您来说都不是很清楚。

有两种方法可以获取更多信息：

1）试用

将qDebug（）或printf（）/ fprintf（）语句放入“慢”线程中的插槽中，然后查看其输出内容。运行几次并得出结论。

2）确保

您将需要阅读源代码，也就是如何使用元对象编译器。moc从源文件获取此信息。这是一项涉及更多的调查，但这可能导致确定性。

据我所知，每个信号发射都会发布一个相应的事件。然后，事件将在线程类中的单独线程中排队。在这里您可以找到相关的两个源代码文件：

无效的QCoreApplication :: postEvent（QObject * Receiver，QEvent * event，int priority）

和

类QPostEventList：公共QVector

它们的权衡有两种方法：

从数据转换器插槽对繁忙的插槽操作进行排队

主要优点是在繁忙操作期间不会丢失信号。但是，这可能会固有地变慢，因为它可能会处理比所需更多的操作。

这个想法是为每个处理的事件重新设置数据，但是真正的繁忙操作仅排队等待执行一次。如果有更多事件，则不一定必须是第一个事件，但这是最简单的实现。

Foo::Foo(QObject *parent) : QObject(parent)
{
    ...
    connect(this, SIGNAL(dataUpdateSignal(const QByteArray&)), SLOT(dataUpdateSlot(const QByteArray&)));
    connect(this, SIGNAL(queueBusyOperationSignal()), SLOT(busyOperation()));
    ...
}

void Foo::dataUpdateSlot(const QByteArray &data)
{
    m_data = data;

    if (busyOperationQueued);
        emit queueBusyOperationSignal();
        m_busyOperationQueued = true;
    }
}

void MyClass::busyOperationSlot()
{

    // Do the busy work with m_data here

    m_busyOperationQueued = false;    
}

连接/断开

这个想法是在开始处理时从相应的信号上断开插槽。这将确保不会捕获到新的信号发射，并在线程有空处理下一个事件时再次将插槽连接到信号。

尽管在连接和下一个偶数处理之间，线程中会有一些空闲时间，但是至少这是实现它的一种简单方法。实际上，根据此处未提供的更多上下文，性能差异甚至可以忽略不计。

主要缺点是，这将在繁忙操作期间丢失信号。

Foo::Foo(QObject *parent) : QObject(parent)
{
    ...
    connect(this, SIGNAL(dataUpdateSignal(const QByteArray&)), SLOT(busyOperationSlot(const QByteArray&)));
    ...
}

void MyClass::busyOperationSlot(const QByteArray &data)
{
    disconnect(this, SIGNAL(dataUpdateSignal(const QByteArray&)), this, SLOT(dataUpdateSlot(const QByteArray&)));

    // Do the busy work with data here

    connect(this, SIGNAL(dataUpdateSignal(const QByteArray&)), SLOT(dataUpdateSlot(const QByteArray&)));
}

未来的想法

我在考虑是否有一个方便的API（例如，类似processEvents（）的方法，但带有仅处理最后发布的事件的参数），用于实际告诉事件系统显式处理最后一个事件，而不是规避问题本身。它看起来确实是这样的API，但是它是私有的。

也许有人会提出功能要求，以在公共场合拥有类似功能。

/*!
\internal
Returns \c true if \a event was compressed away (possibly deleted) and should not be added to the list.
*/
bool QCoreApplication::compressEvent(QEvent *event, QObject *receiver, QPostEventList *postedEvents)

相关的源代码可以在这里找到。

它还似乎在QGuiApplication和中具有替代版本QApplication。

至于完整性，还有这样的方法：

void QCoreApplication :: removePostedEvents（QObject *接收器，int eventType = 0）[静态]

删除使用postEvent（）发送给接收者的给定eventType的所有事件。

事件不会分派，而是从队列中删除。您永远不需要调用此函数。如果确实调用它，请注意终止事件可能会导致接收方破坏一个或多个不变式。

如果receiver为null，则将删除所有对象的eventType事件。如果eventType为0，则将删除所有事件以供接收者使用。永远不要使用eventType为0来调用此函数。如果以这种方式调用它，请注意终止事件可能会导致接收器破坏一个或多个不变式。

但是，根据文档，这不是您想要的。