c++ - Qt 与 QAMQP/RabbitMQ

Question

我正在使用此连接器连接到 RabbitMQ: https://github.com/fuCtor/QAMQP

我的应用程序需要性能和多线程。请问，这个连接器可以使用多线程吗？

我试过:

void Test::newMessage(QAMQP::Queue * q) {
    while (q->hasMessage()) {
        QAMQP::MessagePtr message = q->getMessage();
        MyEvent *me = new MyEvent();
        me->message = message;
        poolThreadPosicao->start(me);
    }
}
class MyEvent : public QRunnable {
    public:
    QAMQP::MessagePtr message;
    void run() {
        s.queue->ack(this->message);
    }
};

在某些消息中，RabbitMQ 会说:“Unacked 10 messages”。 10 条消息是我在代理消息中的 qos。我需要解决这个问题吗？怎么办？

Answer 1

首先我建议你切换到https://github.com/mbroadst/qamqp ，因为它是原始项目(不再处于积极开发中)的替代品。更新后的代码包含许多性能和内存增强功能，以及对 RabbitMQ 的更完整支持。话虽如此，目前该项目的两个版本都旨在为每个线程建立一个连接。这意味着您创建的任何 channel (交换器或队列)都将作为创建它的连接(客户端)的父级，因此绑定(bind)到创建线程。

处理您面临的问题的一种方法是从 QRunnable 和 QObject 继承，在您完成任务时发出消息(注意:这是未经测试的，我只是提供基本结构):

using namespace QAMQP;
class MessageJob : public QRunnable, public QObject
{
    Q_OBJECT
public:
    MessageJob(const Message &message)
        : m_message(message)
    {
    }

    virtual void run() {
        // process the message

        // when you are done, emit the finished signal
        Q_EMIT finished(m_message);
    }

Q_SIGNALS:
    void finished(const Message &message);

private:
    Message m_message;

};

class Test : public QObject
{
    Q_OBJECT
public:
    Test(QObject *parent = 0)
        : QObject(parent)
    {
        // setup and connect client
        // create queue and start consuming
    }

private Q_SLOTS:
    void messageReceived(const Message &message)
    {
        MessageJob *job = new MessageJob; // no parent, this will be autodeleted
        connect(job, SIGNAL(finished(Message)), this, SLOT(jobFinished(Message)), Qt::QueuedConnection);
        // NOTE: Qt::QueuedConnection is very important as it allows the signal to 
        //       cross threads

        QThreadPool::globalInstance()->start(job);
    }

    void jobFinished(const Message &message) {
        m_queue->ack(message);
    }

private:
    Client m_client;
    Queue *m_queue;

};