【知识分享】MQTT实战-使用mosquitto客户端连接emqx服务器

一、简介

    MQTT（Message Queuing Telemetry Transport）是一种轻量级的、基于发布/订阅模式的通信协议，旨在实现物联网设备之间的低带宽、高延迟的通信。MQTT协议设计简洁，使用TCP/IP协议进行通信，适用于各种网络环境，尤其适合在有限的网络带宽和不稳定的网络连接条件下进行通信。

    MQTT的工作原理是基于发布/订阅模式的消息传递，它包括两个主要角色：发布者（Publisher）和订阅者（Subscriber）。发布者负责发布消息到主题（Topic），而订阅者则可以订阅感兴趣的主题，从而接收相应的消息。这种模式使得设备之间可以实现异步通信，同时降低了对网络和系统资源的占用。MQTT的两个常用的开源库，mosquitto和emqx。本文服务端使用的是emqx，而客户端则用mosquitto开源库。

二、通信时序框图

三、服务器搭建

• 创建MQTT服务

    可以进入emqx官网参考下载过程。

    按照官网执行以下三个步骤，即可打开emqx服务。

curl -s https://assets.emqx.com/scripts/install-emqx-deb.sh | sudo bash
sudo apt-get install emqx
sudo systemctl start emqx

    如果要查看服务开启状态，则使用以下命令。

sudo systemctl status emqx

    如果需要关闭服务，则使用以下命令。

sudo systemctl stop emqx

• 服务测试

    首先安装mosquitto客户端。

sudo apt install mosquitto-clients

    先打开一个终端，输入以下命令。

mosquitto_sub -h localhost -t "topic"

    再打开另外一个终端，输入以下命令，如果可以在前一个终端界面上看到发送的信息，则说明服务正常开启。

mosquitto_pub -h localhost -t "topic" -m "Hello, MQTT"

四、客户端代码实现

    这里我们实现两个客户端用来相互通信。
客户端1代码

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <mosquitto.h>

#define MQTT_SERVER_ADDRESS     "127.0.0.1"
#define MQTT_SERVER_PORT        1883
#define MQTT_TOPIC              "topic"

struct mosquitto *mosq = NULL;

void on_connect(struct mosquitto *mosq, void *obj, int rc)
{
    if (rc == 0) {
        printf("Connected to MQTT broker\n");
    } else {
        fprintf(stderr, "Failed to connect to MQTT broker: %s\n", mosquitto_strerror(rc));
        exit(EXIT_FAILURE);
    }
}

void on_publish(struct mosquitto *mosq, void *obj, int mid)
{
    printf("Message published\n");
}

int main()
{
    int rc;
    int protocol_version = MQTT_PROTOCOL_V311;

    mosquitto_lib_init(); // 初始化Mosquitto库

    mosq = mosquitto_new(NULL, true, NULL);
    if (!mosq) {
        fprintf(stderr, "Error: Out of memory.\n");
        return EXIT_FAILURE;
    }

    mosquitto_connect_callback_set(mosq, on_connect);
    mosquitto_publish_callback_set(mosq, on_publish);

    // 设置MQTT协议版本
    mosquitto_opts_set(mosq, MOSQ_OPT_PROTOCOL_VERSION, &protocol_version);
    
    rc = mosquitto_connect(mosq, MQTT_SERVER_ADDRESS, MQTT_SERVER_PORT, 60); // 连接到MQTT broker
    if (rc != MOSQ_ERR_SUCCESS) {
        fprintf(stderr, "Unable to connect to MQTT broker: %s\n", mosquitto_strerror(rc));
        return EXIT_FAILURE;
    }

    printf("Connected to MQTT broker\n");

    while (1) {
        char payload[100];
        printf("Enter message: ");
        fgets(payload, sizeof(payload), stdin);
        payload[strcspn(payload, "\n")] = '\0';

        rc = mosquitto_publish(mosq, NULL, MQTT_TOPIC, strlen(payload), payload, 0, false);
        if (rc != MOSQ_ERR_SUCCESS) {
            fprintf(stderr, "Failed to publish message: %s\n", mosquitto_strerror(rc));
        }
    }

    mosquitto_destroy(mosq); // 销毁Mosquitto对象
    mosquitto_lib_cleanup(); // 清理Mosquitto库

    return 0;
}

客户端2代码

#include <stdio.h>
#include <stdlib.h>
#include <mosquitto.h>

#define MQTT_SERVER_ADDRESS     "localhost"
#define MQTT_SERVER_PORT        1883
#define MQTT_TOPIC              "topic"

struct mosquitto *mosq = NULL;

void on_connect(struct mosquitto *mosq, void *obj, int rc)
{
    if (rc == 0) {
        printf("Connected to MQTT broker\n");
        mosquitto_subscribe(mosq, NULL, MQTT_TOPIC, 0); // 订阅主题
    } else {
        fprintf(stderr, "Failed to connect to MQTT broker: %s\n", mosquitto_strerror(rc));
        exit(EXIT_FAILURE);
    }
}

void on_message(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg)
{
    printf("Received message: %.*s\n", (int)msg->payloadlen, (char *)msg->payload);
}

int main()
{
    int rc;
    int protocol_version = MQTT_PROTOCOL_V311;

    mosquitto_lib_init(); // 初始化Mosquitto库

    mosq = mosquitto_new(NULL, true, NULL);
    if (!mosq) {
        fprintf(stderr, "Error: Out of memory.\n");
        return EXIT_FAILURE;
    }

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

    // 设置MQTT协议版本
    printf("protocol version is:%d\n", protocol_version);
    mosquitto_opts_set(mosq, MOSQ_OPT_PROTOCOL_VERSION, &protocol_version);

    rc = mosquitto_connect(mosq, MQTT_SERVER_ADDRESS, MQTT_SERVER_PORT, 60); // 连接到MQTT broker
    if (rc != MOSQ_ERR_SUCCESS) {
        fprintf(stderr, "Unable to connect to MQTT broker: %s\n", mosquitto_strerror(rc));
        return EXIT_FAILURE;
    }

    printf("Connected to MQTT broker\n");

    mosquitto_loop_start(mosq); // 启动MQTT客户端循环

    while (1) {
        // 保持客户端运行
    }

    mosquitto_disconnect(mosq); // 断开连接
    mosquitto_destroy(mosq); // 销毁Mosquitto对象
    mosquitto_lib_cleanup(); // 清理Mosquitto库

    return 0;
}

• 本地编译测试

gcc -o MyMQTT *.c -lpthread -lmosquitto

五、交叉编译

    先上mosquitto官网下载mosquitto源码，这里下载的是当前最新版本2.0.18。

    解压，并进入文件夹。

tar xvf mosquitto-2.0.18.tar.gz
cd mosquitto-2.0.18

    打开config.mk配置文件，将CJSON注释掉，并且把编译静态库功能打开。

    如果不用SSL加密，把WITH_TLS:=yes也注释掉，不然就得再装个openssl。

    执行make命令进行编译，注意这里要使用交叉编译工具，即把CC替换成交叉编译gcc，把CXX替换成交叉编译g++。

make CC=/opt/atk-dlrk356x-toolchain/bin/aarch64-buildroot-linux-gnu-gcc CXX=/opt/atk-dlrk356x-toolchain/bin/aarch64-buildroot-linux-gnu-g++

    然后再执行安装命令，这里注意加上sudo权限。

sudo make install

    安装后库会在/usr/local/lib中，头文件会在/usr/local/include中。

    然后回到上面的例子工程中，使用交叉编译工具进行编译，编译的时候指定链接静态库mosquitto。

/opt/atk-dlrk356x-toolchain/bin/aarch64-buildroot-linux-gnu-gcc -static -o MyMQTT *.c -lpthread -L/usr/local/lib -lmosquitto -I/usr/local/include

六、注意事项

1、使用mosquitto_connect接口进行MQTT连接时，第4个参数是保持心跳持续时间，这个值不能设为0，不然一直连接不上。
2、注意协议版本号，需要保持通信双方协议版本号一致，正常默认是V3.11。
3、如果要使用交叉编译时，需要安装宿主机对应架构的mosquitto库，比如需要交叉编译arm架构的，那需要手动下载mosquitto的源码包，编译生成对应的库，编译时链接到该库。