c++ nanoMQ 发布或订阅多个消息的程序

小鱼仙官

292人浏览 · 2024-08-22 08:48:56

小鱼仙官 · 2024-08-22 08:48:56 发布

线程方式：

nanomq.h

#ifndef NANOMQ_H
#define NANOMQ_H

#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <iostream>
#include <stdio.h>
#include <MQTTClient.h>

#include <boost/thread.hpp>

#define ADDRESS     "tcp://broker.emqx.io:1883"
#define USERNAME    "xgdd"
#define PASSWORD    "xgdd"
#define CLIENTID    "mqttx_47b97142"
#define QOS         0
#define TIMEOUT     10000L

class NanoMQ
{
public:
    NanoMQ();
    ///
    /// \brief init 连接nanomq
    /// \return  -1-连接失败 0-连接成功
    ///
    int init();
    ///
    /// \brief quit 断开连接
    ///
    void quit();
    ///
    /// \brief sendInfo 发送信息
    /// \param topic    节点
    /// \param data     数据
    ///
    void sendInfo(char *topic, char *data);

private:
    MQTTClient client;

    void publish(MQTTClient client, char *topic, char *payload);

    int receive_subMessage();

    static void receive_subMessage_thread(void* arg);
};


#endif // NANOMQ_H

nanomq.cpp

#include "nanomq.h"

NanoMQ::NanoMQ()
{

}

int NanoMQ::init()
{
//法二
    char* subscribeTopic[] = {"testtopic/123", "testtopic/1234", "testtopic/12345"};
    int m_qos[] = {0, 0, 0};

    MQTTClient_create(&client, ADDRESS, CLIENTID, 0, NULL);
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    conn_opts.username = USERNAME;
    conn_opts.password = PASSWORD;

    if (MQTTClient_connect(client, &conn_opts) != MQTTCLIENT_SUCCESS) {
        printf("Failed to connect\n");
        return -1;
    } else {
        printf("Connected to MQTT Broker!\n");
    }
//法一
//    if (MQTTClient_subscribe(client, "testtopic/123", QOS) != MQTTCLIENT_SUCCESS)
//    {
//        printf("Failed to subscribe\n");
//    }
//    if (MQTTClient_subscribe(client, "testtopic/1234", QOS) != MQTTCLIENT_SUCCESS)
//    {
//        printf("Failed to subscribe\n");
//    }
//    if (MQTTClient_subscribe(client, "testtopic/12345", QOS) != MQTTCLIENT_SUCCESS)
//    {
//        printf("Failed to subscribe\n");
//    }

//法二
    if (MQTTClient_subscribeMany(client, 3, subscribeTopic, m_qos) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to subscribe\n");
    }

    boost::thread t(&NanoMQ::receive_subMessage_thread, this);

    return 0;
}

void NanoMQ::quit()
{
    MQTTClient_disconnect(client, TIMEOUT);
    MQTTClient_destroy(&client);
}

void NanoMQ::sendInfo(char *topic, char *data)
{
    publish(client, topic, data);
}

void NanoMQ::publish(MQTTClient client, char *topic, char *payload)
{
    MQTTClient_message message = MQTTClient_message_initializer;
    message.payload = payload;
    message.payloadlen = strlen(payload);
    message.qos = QOS;
    message.retained = 0;
    MQTTClient_deliveryToken token;
    MQTTClient_publishMessage(client, topic, &message, &token);
    MQTTClient_waitForCompletion(client, token, TIMEOUT);
    printf("Send `%s`\n", payload);
}

int NanoMQ::receive_subMessage()
{
    int topicLen;
    int rc;
    char *topic = "testtopic/123";
    MQTTClient_message *message = NULL;
    while (1) {
        rc = MQTTClient_receive(client, &topic, &topicLen, &message, TIMEOUT);
        if( rc == MQTTCLIENT_SUCCESS && topic!= NULL ){
            printf("Message arrived \n");
            printf("     topic: %s\n", topic);
            printf("   message: %.*s\n", message->payloadlen, (char*)message->payload);
        }
        printf("MQTTClient receive message, return code %d\n", rc);

        sleep(1);
    }

    return rc;
}

void NanoMQ::receive_subMessage_thread(void *arg)
{
    NanoMQ *m_nanomq = (NanoMQ *)arg;
    m_nanomq->receive_subMessage();
}

main.cpp

#include <iostream>
#include <thread>

#include "nanomq.h"

using namespace std;

NanoMQ nanomq;

void topic_1()
{
    while(true) {
        nanomq.sendInfo("testtopic/1", "this is test topic 1");
        sleep(1);
    }
}

void topic_2()
{
    while(true) {
        nanomq.sendInfo("testtopic/2", "this is test topic 2");
        sleep(1);
    }
}

int main(int argc, char *argv[])
{
    cout<<"hello world"<<endl;

    if(nanomq.init() < 0) {
        cout<<"nanomq connect failed.."<<endl;
    }

    thread t1(&topic_1);
    thread t2(&topic_2);

    while (true) {
        sleep(1);
    }

    nanomq.quit();

    return 0;
}

回调函数方式：

nanomq.h

#ifndef NANOMQ_H
#define NANOMQ_H

#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <iostream>
#include <stdio.h>
#include <MQTTClient.h>

#include <boost/thread.hpp>

#define ADDRESS     "tcp://broker.emqx.io:1883"
#define USERNAME    "xgdd"
#define PASSWORD    "xgdd"
#define CLIENTID    "mqttx_47b97149"
#define QOS         0
#define TIMEOUT     10000L

using namespace std;

class NanoMQ
{
public:
    NanoMQ();
    ///
    /// \brief init 连接nanomq
    /// \return  -1-连接失败 0-连接成功
    ///
    int init();
    ///
    /// \brief quit 断开连接
    ///
    void quit();
    ///
    /// \brief sendInfo 发送信息
    /// \param topic    节点
    /// \param data     数据
    ///
    void sendInfo(char *topic, char *data);

private:
    MQTTClient client;

    void publish(MQTTClient client, char *topic, char *payload);
};

///
/// \brief on_message   接收消息
/// \param context
/// \param topicName    节点 用于区分不同节点
/// \param topicLen
/// \param message      数据 结构中的message->payload
/// \return
///
int on_message(void *context, char *topicName, int topicLen, MQTTClient_message *message);

#endif // NANOMQ_H

namomq.cpp

#include "nanomq.h"

int on_message(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    printf("Message arrived \n");
    printf("  topic: %s\n", topicName);
    printf("  message: %d...%s\n", message->payloadlen, (char*)message->payload);

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);

    return 1;
}

NanoMQ::NanoMQ()
{

}

int NanoMQ::init()
{
    char* subscribeTopic[] = {"testtopic/123", "testtopic/1234", "testtopic/12345"};
    int m_qos[] = {0, 0, 0};

    MQTTClient_create(&client, ADDRESS, CLIENTID, 0, NULL);
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    conn_opts.username = USERNAME;
    conn_opts.password = PASSWORD;

    MQTTClient_setCallbacks(client, NULL, NULL, on_message, NULL);

    if (MQTTClient_connect(client, &conn_opts) != MQTTCLIENT_SUCCESS) {
        printf("Failed to connect\n");
        return -1;
    } else {
        printf("Connected to MQTT Broker!\n");
    }

    if (MQTTClient_subscribeMany(client, 3, subscribeTopic, m_qos) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to subscribe\n");
    }

    return 0;
}

void NanoMQ::quit()
{
    MQTTClient_disconnect(client, TIMEOUT);
    MQTTClient_destroy(&client);
}

void NanoMQ::sendInfo(char *topic, char *data)
{
    MQTTClient_subscribe(client, topic, QOS);

    publish(client, topic, data);
}

void NanoMQ::publish(MQTTClient client, char *topic, char *payload)
{
    MQTTClient_message message = MQTTClient_message_initializer;
    message.payload = payload;
    message.payloadlen = strlen(payload);
    message.qos = QOS;
    message.retained = 0;
    MQTTClient_deliveryToken token;
    MQTTClient_publishMessage(client, topic, &message, &token);
    MQTTClient_waitForCompletion(client, token, TIMEOUT);
//    printf("Send `%s`\n", payload);
}

mian.cpp

内容不变

腾讯云开发者社区

腾讯云面向开发者汇聚海量精品云计算使用和开发经验，营造开放的云计算技术生态圈。

更多推荐

Elasticsearch复杂数据类型终极指南：从入门到精通

Elasticsearch作为功能强大的搜索引擎，支持多种复杂数据类型，让开发者能够灵活处理各种结构化和非结构化数据。本文将带你全面了解Elasticsearch中的复杂数据类型，从基础概念到实际应用，助你轻松掌握数据建模的核心技巧。## 内部对象：构建层级化数据结构在Elasticsearch中，对象类型（Object）是最基础的复杂数据类型之一，用于表示具有嵌套关系的数据。例如，我们可

腾讯云开发者社区

终极指南：Flink SQL连接器版本管理从混乱到有序的升级之路

Apache Flink作为流处理领域的佼佼者，其SQL连接器的版本管理一直是开发者面临的核心挑战。本文将系统讲解Flink SQL连接器版本管理的最佳实践，帮助你轻松应对版本兼容性问题，实现从混乱到有序的升级之旅。## 连接器版本管理的常见痛点 😫在Flink应用开发中，连接器版本管理常常让开发者头疼不已。不同版本的连接器可能导致各种兼容性问题，例如API变更、功能差异甚至运行时错误。

腾讯云开发者社区

如何快速搭建Neon无服务器PostgreSQL：面向初学者的完整指南

Neon是一款革命性的无服务器PostgreSQL解决方案，它通过分离存储和计算层，实现了自动扩缩容、类代码式数据库分支以及零级扩展能力。本指南将帮助你从零开始搭建Neon开发环境，体验这款创新数据库的强大功能。## 准备工作：环境要求与依赖项在开始搭建Neon环境前，请确保你的系统满足以下要求：- Linux操作系统（推荐Ubuntu 20.04+或Debian 11+）- Git