当前位置：首页 > news >正文

Qt-ZMQ的使用补充（pub-sub）

news 来源：原创 2025/5/14 20:02:00

之前写过一篇Qt使用ZMQ的博客Qt网络编程-ZMQ的使用，本文是其的补充部分。

Linux上编译使用

首先这次实在Linux上进行演示，下载zmq源码，安装cmake，使用cmake进行编译。下载之后解压：

输入命令：

cd ..

mkdir zmqbuild

cmake ../libzmq-master

会有以下输出：

输入：

make -j16

等待编译完成：

在该目录下即可找到对应的动态库文件：

新建工程，然后将头文件和动态库文件放进来，然后在工程文件中引入头文件和库文件。

这是我的文件结构：

工程文件：


#引入libzmq头文件和库
INCLUDEPATH += $PWD/zmq/include


unix {
    LIBS += -L$$PWD/zmq/lib
    LIBS += -lzmq
}

基于TCP通信

在之前的博客中发布和订阅段有对应的bind和connect方法进行绑定和连接：

bool ZmqPublisher::bind(quint16 port) {
    QString arg = QString("tcp://*:%1").arg(port);
    int rc = zmq_bind(socket, arg.toUtf8().constData());
    return rc == 0;
}

bool ZmqSubscriber::connectTo(quint16 port) {
    QString arg = QString("tcp://localhost:%1").arg(port);
    int rc = zmq_connect(socket, arg.toUtf8().constData());
    return rc == 0;
}

这两个方法内部使用tcp进行网络通信。bind和connect也可以绑定具体的ip和端口，以下是扩充方法：

bool ZmqPublisher::bind(const QString &ip, quint16 port)
{
    QString arg = QString("tcp://%1:%2").arg(ip).arg(port);
    int rc = zmq_bind(socket, arg.toUtf8().constData());
    return rc == 0;
}

bool ZmqSubscriber::connectTo(const QString &ip, quint16 port)
{
    QString arg = QString("tcp://%1:%2").arg(ip).arg(port);
    int rc = zmq_connect(socket, arg.toUtf8().constData());
    return rc == 0;
}

写一个测试程序进行测试：

可以看到发布端bind后对应的tcp开始listen了。

然后订阅段进行connect：

对应的tcp连接已经建立。

测试通信：

通信正常。

基于本地文件通信

zmq除了借助TCP网络进行进程间通信还可以使用文件进行进程间通信，这里封装对应的方法：

bool ZmqPublisher::bind(const QString &path)
{
    int rc = zmq_bind(socket, path.toUtf8().constData());
    return rc == 0;
}

bool ZmqSubscriber::connectTo(const QString &path)
{
    int rc = zmq_connect(socket, path.toUtf8().constData());
    return rc == 0;
}

直接传入对应的地址，当然这里path传“tcp://{ip}:{port}”这种也行，这样就是前面提到的网络通信了。本地文件通信传入格式是“ipc://{path}”。

可以看到生成了一个文件：

使用命令查看文件属性：

这就是进行通信的socket文件。订阅端进行connect：

使用命令查看连接情况，之前在TCP和UDP通信的博客中讲到了使用netstat查看网络连接，实际上netstat也可以查看这种本地文件连接：

使用ss命令也可以查看，ss命令功能比netstat命令广泛，后续推荐使用ss命令。

通信测试：

注意事项：

1.注意对应文件的路径，因为我的示例中socket文件在执行文件同级目录下所以使用的是相对路径。（bind和connect时传入完整路径：比如bind("/home/pc/zkh/projects/build-zmqproject-Desktop_Qt_5_12_12_GCC_64bit-Debug/test")）

2.直接使用本地文件bind(ipc://{path})这种方式通信目前不能跨主机而且在windows上无效。

代码优化

之前写的代码调用比较麻烦，其中要在调用处新建线程类，然后订阅部分每connect一个都要新建sub对象，所以我在pub和sub类之上在新建一个类ZmqPubSub，对二者进行管理，调用者只需要新建ZmqPubSub这一个类就可以进行发布订阅消息处理相关操作，而且方便处理多个订阅端，以下是代码：

头文件：

#ifndef ZMQPUBSUB_H
#define ZMQPUBSUB_H

#include <QObject>

class QThread;
class ZmqSubscriber;
class ZmqPublisher;
class ZmqPubSub : public QObject
{
    Q_OBJECT
public:
    explicit ZmqPubSub(QObject *parent = nullptr);
    void init(const QString &pubPath,const QStringList &subPaths);

signals:
    void dataReceived(const QByteArray &data);

public slots:
    void publishData(const QByteArray &data);

private:
    bool m_Inited;
    ZmqPublisher *m_Pub;
    QList<ZmqSubscriber *>m_Subs;
    QList<QThread *>m_Ths;

};

#endif // ZMQPUBSUB_H

源文件：

#include "zmqpubsub.h"
#include "zmqpublisher.h"
#include "zmqsubscriber.h"
#include <QThread>
#include <QDataStream>

ZmqPubSub::ZmqPubSub(QObject *parent) : QObject(parent),m_Inited(false),m_Pub(nullptr)
{

}

void ZmqPubSub::init(const QString &pubPath, const QStringList &subPaths)
{
    if(m_Inited)return;
    QThread *pubTh=new QThread;
    m_Pub=new ZmqPublisher;
    m_Pub->bind(pubPath);
    m_Pub->moveToThread(pubTh);
    pubTh->start();
    m_Ths.append(pubTh);
    for(const QString &subPath:subPaths)
    {
        QThread *subTh=new QThread;
        ZmqSubscriber *sub=new ZmqSubscriber;
        sub->connectTo(subPath);
        sub->moveToThread(subTh);
        subTh->start();
        m_Subs.append(sub);
        m_Ths.append(subTh);
        connect(sub, &ZmqSubscriber::dataReceived,this,&ZmqPubSub::dataReceived);
        QMetaObject::invokeMethod(sub, &ZmqSubscriber::procesMessage);
    }

    m_Inited=true;
}

void ZmqPubSub::publishData(const QByteArray &data)
{
    if(m_Inited)
    {
       m_Pub->publishData(data);
    }
}

调用的时候只需要新建ZmqPubSub类即可，线程相关在内部自行处理，init方法初始化可以处理多个订阅端的问题。一个类实现发布订阅功能。