带有 ZeroMQ 发布订阅模式的 Qt

Question

我想将 ZeroMQ(4.1.2) 与 Qt (5.2.1) 一起使用。 想法是拥有 zmq pub/sub（服务器在外面），sub 是 qt app。 目前在 Qt 应用程序中运行一次，有人可以提示吗？ ZeroMQ 接收器是否应该以其他方式实现？

目前我的代码如下：

ma​​inwindow.h

namespace Ui {
class MainWindow;
}

class MainWindow : public QMainWindow
{
Q_OBJECT

public:
    explicit MainWindow(QWidget *parent = 0);
    ~MainWindow();

private slots:
    void on_pushButton_clicked();

    void readZMQData();

private:
    Ui::MainWindow *ui;
    QSocketNotifier *qsn;
    void *context;
    void *subscriber;

};

ma​​inwindow.cpp

MainWindow::MainWindow(QWidget *parent) :
    QMainWindow(parent),
    ui(new Ui::MainWindow)
{
    ui->setupUi(this);

    /***** ZMQ *****/

    context = zmq_ctx_new ();
    subscriber = zmq_socket (context, ZMQ_SUB);
    int rc = zmq_connect (subscriber, "tcp://localhost:5556");

    char *filter = "";
    rc = zmq_setsockopt (subscriber, ZMQ_SUBSCRIBE,filter, strlen (filter));
    unsigned int fd=0;
    size_t fd_size = sizeof(fd);
    rc = zmq_getsockopt(subscriber,ZMQ_FD,&fd,&fd_size);

    qsn = new QSocketNotifier(fd, QSocketNotifier::Read, this);
    connect(qsn, SIGNAL(activated(int)), this, SLOT(readZMQData()), Qt::DirectConnection);

}

MainWindow::~MainWindow()
{
    zmq_close (this->subscriber);
    zmq_ctx_destroy (this->context);
    delete ui;
}


void MainWindow::readZMQData()
{
    qsn->setEnabled(false);
    qDebug() << "Got data!";

    int events = 0;
    std::size_t eventsSize = sizeof(events);
    zmq_getsockopt(subscriber,ZMQ_EVENTS, &events, &eventsSize);
    if(events & ZMQ_POLLIN){
        qDebug() << " ======  Data to read ======";

        char *string = s_recv(subscriber);
        qDebug() << "DATA: " << string;
        free(string);
    }

    qsn->setEnabled(true);
}

服务器应用是（来自 ZeroMQ 示例）：

#include "zhelpers.h"

int main (void)
{
    //  Prepare our context and publisher
    void *context = zmq_ctx_new ();
    void *publisher = zmq_socket (context, ZMQ_PUB);
    int rc = zmq_bind (publisher, "tcp://*:5556");
    assert (rc == 0);

    //  Initialize random number generator
    srandom ((unsigned) time (NULL));
    while (1) {
        //  Get values that will fool the boss
        int zipcode, temperature, relhumidity;
        zipcode     = randof (100000);
        temperature = randof (215) - 80;
        relhumidity = randof (50) + 10;

        //  Send message to all subscribers
        char update [20];
        sprintf (update, "%05d %d %d", zipcode, temperature, relhumidity);
        s_send (publisher, update);
    }
    zmq_close (publisher);
    zmq_ctx_destroy (context);
    return 0;
}

Answer 1

套接字通知器看起来应该可以工作。您是否阅读过有关正确处理它的文档？特别是如果您在 Windows 上，在执行读取时似乎有特殊的方法来处理它...禁用、读取等。

http://doc.qt.io/qt-5/qsocketnotifier.html#details

希望对您有所帮助。

Answer 2

第一个提供帮助的 tnx，

我发现了问题，当 ZeroMQ 通知有要阅读的消息时，您需要阅读所有消息，而不仅仅是第一个。

    void MainWindow::readZMQData(int fd)
{
    qsn->setEnabled(false);

    int events = 0;
    std::size_t eventsSize = sizeof(events);
    zmq_getsockopt(subscriber,ZMQ_EVENTS, &events, &eventsSize);
    if(events & ZMQ_POLLIN){
        qDebug() << " ======  Data to read ======";

        char *string;
        // THIS IS THE TRICK! READ UNTIL THERE IS MSG
        while((string = s_recv_nb(subscriber)) != NULL){
            qDebug() << "DATA: " << string;
            free(string);
        }
    }

    qsn->setEnabled(true);
}

Answer 3

由于不清楚 s_recv_nb(zmq::socket_t & socket) 是什么，我将提供我的 - 稍微详细一点 - 实现：

      void MainWindow::readZMQData()
      {
       qsn->setEnabled(false);

       int events = 0;
       std::size_t eventsSize = sizeof(events);
       zmq_getsockopt(subscriber,ZMQ_EVENTS, &events, &eventsSize);
       if(events & ZMQ_POLLIN){
       qDebug() << " ======  Data to read ======";

       char *string;
       int64_t more;
       size_t more_size = sizeof (more);

       do {
           /* Create an empty ØMQ message to hold the message part */
           zmq_msg_t part;
           int rc = zmq_msg_init (&part);
           assert (rc == 0);
           rc = zmq_msg_recv (&part, subscriber, 0);
           assert (rc != -1);
           /* Determine if more message parts are to follow */
           rc = zmq_getsockopt (subscriber, ZMQ_RCVMORE, &more, &more_size);
           assert (rc == 0);
           string = (char*) zmq_msg_data(&part);
           qDebug() << QString(string) ; // << "more" << more;

           zmq_msg_close (&part); 
          } while (more);
       }

       qsn->setEnabled(true);
     }

还有一点：在 Windows 64 的情况下，文件描述符 fd 应该是 uint64_t，就像在 zmq_getsockopt returns EINVAL on windows x64 when local address of ZMQ_FD option_val passed 中一样