基于C++在windows下使用mosquitto实现MQTT订阅客户端

1、MOSQUITTO订阅客户端搭建

#include <iostream>
#include <iostream>
#include <thread>
#include <mosquitto.h>

//mqtt
#define MQTT_USER_NAME "admin"
#define MQTT_PASS_WORD "123456"
#define MQTT_SERVER_IP_ADDRESS "10.57.6.250"
#define MQTT_SERVER_PORT 1883
#define PUBLISH_JSON_DATA_ORG "4321"
static struct mosquitto* mosq;
static bool mqtt_init_sign;
//mqtt

void on_message(struct mosquitto* mosq, void* userdata, const struct mosquitto_message* msg) {
    std::cout << "\nrev msg:" << std::endl;
    std::cout << "the topic of the msg is: " << msg->topic << std::endl;

    if (msg->payloadlen > 0) {
        std::string message(static_cast<char*>(msg->payload), msg->payloadlen);
        std::cout << "the msg content is: " << message.c_str() << std::endl;
    }
    else {
        std::cout << "the msg content is:null" << std::endl;
    }

    std::cout << "QoS: " << msg->qos << std::endl;
    std::cout << "the ID of the msg is: " << msg->mid << std::endl;
    std::cout << "---------------------" << std::endl;
}

void on_connect(struct mosquitto *mosq, void *obj, int rc) {
    if (rc == 0) {
        printf("authentication success\n");
    }
    else {
        printf("connect failed: %s\n", mosquitto_strerror(rc));
    }
}

void on_disconnect(struct mosquitto *mosq, void *obj, int rc) {
    printf("disconnect: %s\n", mosquitto_strerror(rc));

    if (rc == MOSQ_ERR_AUTH) {
        printf("error：authentication failure(username/password is wrong)\n");
    }
    else if (rc == MOSQ_ERR_CONN_REFUSED) {
        printf("error：access refused\n");
    }
}

int init_mqtt(const std::string username, const std::string passwd, const std::string hostname, const int port)
{
    mosquitto_lib_init();

    mosq = mosquitto_new(nullptr, true, nullptr);//第一个形参制定为一个值后，派生类会出现发布数据失败的情况，因为不同的派生类共用一个client_id
    if (!mosq) {
        std::cerr << "Failed to create mosquitto instance!" << std::endl;
        return 0;
    }

    mosquitto_connect_callback_set(mosq, on_connect);
    mosquitto_disconnect_callback_set(mosq, on_disconnect);
    mosquitto_message_callback_set(mosq, on_message);

    const char* usrname = username.c_str();
    const char* password = passwd.c_str();

    int rc = mosquitto_username_pw_set(mosq, usrname, password);
    if (rc != MOSQ_ERR_SUCCESS) {
        std::cerr << "Failed to set username/password: " << mosquitto_strerror(rc) << std::endl;
        mosquitto_destroy(mosq);
        mosquitto_lib_cleanup();
        return 0;
    }

    rc = mosquitto_connect(mosq, hostname.c_str(), port, 60);
    if (rc != MOSQ_ERR_SUCCESS) {
        std::cerr << "Connection failed: " << mosquitto_strerror(rc) << std::endl;

        std::cout << "Trying without authentication..." << std::endl;
        mosquitto_username_pw_set(mosq, nullptr, nullptr);

        rc = mosquitto_connect(mosq, hostname.c_str(), port, 60);
        if (rc != MOSQ_ERR_SUCCESS) {
            std::cerr << "Still failed: " << mosquitto_strerror(rc) << std::endl;
            mosquitto_destroy(mosq);
            mosquitto_lib_cleanup();
            return 0;
        }
    }

    mosquitto_loop_start(mosq);
    mqtt_init_sign = true;
    return 1;
}

void subscribe_msg_from_server()
{
    const char* topic = "TestMqttTopic";
    int qos = 1;  // 服务质量等级 0,1,2

    int result = mosquitto_subscribe(mosq, NULL, topic, qos);
    if (result != MOSQ_ERR_SUCCESS) {
        std::cerr << "订阅失败: " << mosquitto_strerror(result) << std::endl;
    }
    else {
        std::cout << "已订阅主题: " << topic << std::endl;
    }
}

int main()
{
    init_mqtt(MQTT_USER_NAME, MQTT_PASS_WORD, MQTT_SERVER_IP_ADDRESS, MQTT_SERVER_PORT);

    while (1)
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(5000));
        std::cout << "start subscribe!" << std::endl;
        subscribe_msg_from_server();
    }
}

2、效果展示