MQTT在linux下服务端和客户端的应用

MQTT（Message Queuing Telemetry Transport）是一种轻量级、开放标准的消息传输协议，设计用于受限设备和低带宽、不稳定网络的通信。

MQTT的一些关键特点和概念：

发布/订阅模型：

MQTT采用发布/订阅（Publish/Subscribe）模型。发布者（Publisher）发布消息到特定的主题（Topic），而订阅者（Subscriber）则订阅感兴趣的主题。这种模型提供了一种松耦合的通信方式。MQTT支持基于TLS/SSL的加密通信，提供一定的安全性。

主题（Topic）：

主题是消息的标签或类别。发布者将消息发布到特定的主题，而订阅者则通过订阅特定的主题来接收相关消息。

消息队列：

MQTT消息以队列的方式传输。当发布者发布消息时，消息被发送到相应主题的队列，订阅者可以从队列中接收消息。

QoS级别（Quality of Service）：

MQTT支持不同的QoS级别，用于确保消息的可靠性和传递顺序。包括：
QoS 0：最多一次，消息发出后不进行确认。
QoS 1：至少一次，确保消息至少被传递一次。
QoS 2：只有一次，确保消息仅被传递一次。

保持连接：

MQTT客户端与服务器之间保持持久连接。这允许客户端随时接收实时消息，而不需要频繁地建立和断开连接。

清理会话（Clean Session）：

客户端可以选择创建一个清理会话或一个持久会话。清理会话意味着服务器不会保留客户端的订阅信息，而持久会话则会保存订阅信息。

遗嘱消息（Will Message）：

客户端可以指定遗嘱消息，以便在客户端异常断开连接时，服务器可以将遗嘱消息发布到预定的主题。

中继器（Broker）：

MQTT系统通常包含一个中继器或代理，负责接收、路由和分发消息。这个中继器被称为MQTT Broker。

MQTT服务端

首先，安装MQTT库，例如 Eclipse Paho MQTT C库。

// mqtt_server.c

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

#define SERVER_ADDRESS "tcp://localhost:1883"
#define CLIENT_ID "MQTT_Server"
#define TOPIC "test_topic"

volatile MQTTClient_deliveryToken deliveredtoken;

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

    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
}

int main() {
   
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;

    // Create MQTT client
    MQTTClient_create(&client, SERVER_ADDRESS, CLIENT_ID, MQTTCLIENT_PERSISTENCE_NONE, NULL);

    // Set callback for incoming messages
    MQTTClient_setCallbacks(client, NULL, NULL, messageArrived, NULL);

    // Connect to the broker
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS) {
   
        printf("Failed to connect to the broker, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }

    // Subscribe to a topic
    if ((rc = MQTTClient_subscribe(client, TOPIC, 1)) != MQTTCLIENT_SUCCESS) {
   
        printf("Failed to subscribe to topic, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }

    printf("MQTT server is running...\n");

    // Keep the program running
    while (1) {
   
        sleep(1);
    }

    // Disconnect from the broker
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);

    return 0;
}

MQTT客户端

// mqtt_client.c

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

#define SERVER_ADDRESS "tcp://localhost:1883"
#define CLIENT_ID "MQTT_Client"
#define TOPIC "test_topic"
#define QOS 1

int main() {
   
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;

    // Create MQTT client
    MQTTClient_create(&client, SERVER_ADDRESS, CLIENT_ID, MQTTCLIENT_PERSISTENCE_NONE, NULL);

    // Connect to the broker
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS) {
   
        printf("Failed to connect to the broker, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }

    // Publish a message
    const char *message = "Hello, MQTT!";
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    pubmsg.payload = (void *)message;
    pubmsg.payloadlen = (int)strlen(message);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    MQTTClient_deliveryToken token;
    if ((rc = MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token)) != MQTTCLIENT_SUCCESS) {
   
        printf("Failed to publish message, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }

    printf("Message published to topic: %s\n", TOPIC);

    // Disconnect from the broker
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);

    return 0;
}