MQTT 简介

MQTT（Message Queuing Telemetry Transport） 是一种基于发布/订阅模式的轻量级消息传输协议，专为低带宽、高延迟或不稳定的网络环境设计。广泛应用于物联网（IoT）、移动应用、智能家居等领域。

MQTT 的核心作用与功能

  1. 轻量高效：协议开销小，适合资源受限的设备。
  2. 发布/订阅模型：
    • 发布者（Publisher）：发送消息到特定主题（Topic）。
    • 订阅者（Subscriber）：订阅主题以接收消息。
    • 代理服务器（Broker）：中转消息，负责主题匹配与消息分发。
  3. 服务质量（QoS）：
    • QoS 0：最多一次（可能丢失）。
    • QoS 1：至少一次（可能重复）。
    • QoS 2：恰好一次（可靠传输）。
  4. 遗嘱消息（LWT）：客户端异常断开时，代理自动发布预设消息。
  5. 保留消息（Retained）：新订阅者立即收到主题的最新消息。

MQTT 通信流程

  1. 客户端连接代理：
    CONNECT
    CONNACK
    Client
    Broker
  2. 订阅主题：
    SUBSCRIBE
    SUBACK
    Subscriber
    Broker
  3. 消息发布与接收：
    PUBLISH
    PUBLISH
    Publisher
    Broker
    Subscriber
  4. 断开连接：
    DISCONNECT
    Client
    Broker

MQTT 工作原理

  • 主题分层：主题以 / 分隔，如 sensors/temperature/room1。
  • 通配符：
    • +：单层匹配（如 sensors/+/room1）。
    • #：多层匹配（如 sensors/#）。
  • 代理路由：代理根据主题匹配将消息分发给订阅者。

在Ubuntu上下载、编译和使用

在Ubuntu系统中，我们将使用两种库实现MQTT通信：paho_mqtt.c（Eclipse Paho C客户端库）和mosquitto（Mosquitto库）。首先安装代理和依赖：

  1. 安装依赖
sudo apt update
sudo apt install build-essential cmake libssl-dev

  1. 安装Mosquitto Broker：
    Mosquitto是一个流行的MQTT代理，用于测试。在终端执行：

    sudo apt update
sudo apt install mosquitto mosquitto-clients
sudo apt install libmosquitto-dev

    启动代理：

    sudo systemctl start mosquitto

    验证代理运行：sudo systemctl status mosquitto。

  2. 安装 Paho MQTT C 库：

    • 通用依赖：sudo apt install build-essential cmake openssl libssl-dev
    • paho_mqtt.c库：从源码编译安装。
      git clone https://github.com/eclipse/paho.mqtt.c.git
cd paho.mqtt.c
cmake -B build -DCMAKE_INSTALL_PREFIX=/usr/local
cmake --build build
sudo cmake --install build

现在，我们可以编写C语言代码实现客户端和服务端通信。服务端在MQTT上下文中通常指订阅者客户端，我们将实现：

  • 客户端：发布消息。
  • 服务端：订阅消息。
    使用两种库分别实现，并添加TLS支持。

使用paho_mqtt.c库实现基本通信

paho_mqtt.c提供简单API。我们将编写一个发布者客户端和一个订阅者服务端。

主要函数介绍（需检查返回值）：

  • MQTTClient_create()：创建MQTT客户端实例。参数：客户端句柄指针、代理地址、客户端ID、持久存储类型。
  • MQTTClient_connect()：连接到代理。参数：客户端句柄、连接选项（包括用户名、密码等）。
  • MQTTClient_publish()：发布消息到主题。参数：客户端句柄、主题、消息长度、消息内容、QoS、保留标志。
  • MQTTClient_subscribe()：订阅主题。参数：客户端句柄、主题、QoS。
  • MQTTClient_setCallbacks()：设置消息到达回调函数。
  • MQTTClient_disconnect()：断开连接。
  • MQTTClient_destroy()：销毁客户端。

每个函数返回int类型，成功时返回MQTTCLIENT_SUCCESS（0），失败返回错误码。必须检查返回值处理错误。

代码实现：

  1. 发布者客户端（publisher_paho.c）：
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <MQTTClient.h>
#include <unistd.h>

#define ADDRESS "tcp://localhost:1883" // 代理地址
#define CLIENTID "PublisherClient"
#define TOPIC "test/topic"
#define QOS 1
#define TIMEOUT 10000L

int main() {
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;
    int rc;

    // 创建客户端
    rc = MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    if (rc != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "Failed to create client: %d\n", rc);
        exit(1);
    }

    // 设置连接选项
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    // 连接到代理
    rc = MQTTClient_connect(client, &conn_opts);
    if (rc != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "Failed to connect: %d\n", rc);
        MQTTClient_destroy(&client);
        exit(1);
    }

    // 发布消息
    pubmsg.payload = "Hello from Paho Publisher";
    pubmsg.payloadlen = strlen(pubmsg.payload);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    rc = MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token);
    if (rc != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "Failed to publish: %d\n", rc);
        MQTTClient_disconnect(client, TIMEOUT);
        MQTTClient_destroy(&client);
        exit(1);
    }

    printf("Message published. Waiting for delivery token...\n");
    MQTTClient_waitForCompletion(client, token, TIMEOUT);

    // 断开并清理
    rc = MQTTClient_disconnect(client, TIMEOUT);
    if (rc != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "Failed to disconnect: %d\n", rc);
    }
    MQTTClient_destroy(&client);
    return 0;
}
  1. 订阅者服务端（subscriber_paho.c）：
#include <stdio.h>
#include <stdlib.h>
#include <MQTTClient.h>
#include <unistd.h>

#define ADDRESS "tcp://localhost:1883"
#define CLIENTID "SubscriberClient"
#define TOPIC "test/topic"
#define QOS 1
#define TIMEOUT 10000L

int messageArrived(void *context, char *topicName, int topicLen, MQTTClient_message *message) {
    printf("Message arrived on topic: %s\n", topicName);
    printf("Message: %.*s\n", message->payloadlen, (char *)message->payload);
    MQTTClient_free(topicName);
    MQTTClient_freeMessage(&message);
    return 1;
}

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

    rc = MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    if (rc != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "Failed to create client: %d\n", rc);
        exit(1);
    }

    // 设置回调
    rc = MQTTClient_setCallbacks(client, NULL, NULL, messageArrived, NULL);
    if (rc != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "Failed to set callbacks: %d\n", rc);
        MQTTClient_destroy(&client);
        exit(1);
    }

    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    rc = MQTTClient_connect(client, &conn_opts);
    if (rc != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "Failed to connect: %d\n", rc);
        MQTTClient_destroy(&client);
        exit(1);
    }

    // 订阅主题
    rc = MQTTClient_subscribe(client, TOPIC, QOS);
    if (rc != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "Failed to subscribe: %d\n", rc);
        MQTTClient_disconnect(client, TIMEOUT);
        MQTTClient_destroy(&client);
        exit(1);
    }

    printf("Subscribed to topic: %s. Waiting for messages...\n", TOPIC);
    while(1) {
        // 保持运行以接收消息
        sleep(1);
    }

    // 清理（实际中需处理退出信号）
    rc = MQTTClient_unsubscribe(client, TOPIC);
    if (rc != MQTTCLIENT_SUCCESS) fprintf(stderr, "Unsubscribe failed: %d\n", rc);
    rc = MQTTClient_disconnect(client, TIMEOUT);
    if (rc != MQTTCLIENT_SUCCESS) fprintf(stderr, "Disconnect failed: %d\n", rc);
    MQTTClient_destroy(&client);
    return 0;
}

编译和运行测试：

  • 编译发布者：
    gcc publisher_paho.c -o publisher_paho -lpaho-mqtt3c
  • 编译订阅者：
    gcc subscriber_paho.c -o subscriber_paho -lpaho-mqtt3c
  • 运行测试：
    1. 启动订阅者：./subscriber_paho（后台运行）。
    2. 启动发布者：./publisher_paho。
  • 输出结果：
    • 发布者输出：Message published. Waiting for delivery token...
    • 订阅者输出：Message arrived on topic: test/topic 和 Message: Hello from Paho Publisher

使用mosquitto库实现基本通信

mosquitto库提供类似功能。我们将编写发布者和订阅者。

主要函数介绍（需检查返回值）：

  • mosquitto_new()：创建mosquitto实例。参数：客户端ID、clean session标志、用户数据。
  • mosquitto_connect()：连接到代理。参数：实例、主机、端口、保持连接时间。
  • mosquitto_publish()：发布消息。参数：实例、消息ID指针、主题、消息长度、消息内容、QoS、保留标志。
  • mosquitto_subscribe()：订阅主题。参数：实例、消息ID指针、主题、QoS。
  • mosquitto_message_callback_set()：设置消息回调函数。
  • mosquitto_disconnect()：断开连接。
  • mosquitto_destroy()：销毁实例。
  • mosquitto_loop_start()：启动网络循环线程。

函数返回int，成功时返回MOSQ_ERR_SUCCESS（0），失败返回错误码。

代码实现：

  1. 发布者客户端（publisher_mosquitto.c）：
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <mosquitto.h>
#include <unistd.h>

#define HOST "localhost"
#define PORT 1883
#define TOPIC "test/topic"
#define QOS 1
#define MESSAGE "Hello from Mosquitto Publisher"

int main() {
    struct mosquitto *mosq = NULL;
    int rc;

    // 初始化库
    mosquitto_lib_init();

    // 创建实例
    mosq = mosquitto_new(NULL, true, NULL);
    if (!mosq) {
        fprintf(stderr, "Failed to create client\n");
        mosquitto_lib_cleanup();
        return 1;
    }

    // 连接到代理
    rc = mosquitto_connect(mosq, HOST, PORT, 60);
    if (rc != MOSQ_ERR_SUCCESS) {
        fprintf(stderr, "Failed to connect: %s\n", mosquitto_strerror(rc));
        mosquitto_destroy(mosq);
        mosquitto_lib_cleanup();
        return 1;
    }

    // 发布消息
    rc = mosquitto_publish(mosq, NULL, TOPIC, strlen(MESSAGE), MESSAGE, QOS, false);
    if (rc != MOSQ_ERR_SUCCESS) {
        fprintf(stderr, "Failed to publish: %s\n", mosquitto_strerror(rc));
        mosquitto_disconnect(mosq);
        mosquitto_destroy(mosq);
        mosquitto_lib_cleanup();
        return 1;
    }

    printf("Message published\n");

    // 断开并清理
    mosquitto_disconnect(mosq);
    mosquitto_destroy(mosq);
    mosquitto_lib_cleanup();
    return 0;
}
  1. 订阅者服务端（subscriber_mosquitto.c）：
#include <stdio.h>
#include <stdlib.h>
#include <mosquitto.h>
#include <unistd.h>

#define HOST "localhost"
#define PORT 1883
#define TOPIC "test/topic"
#define QOS 1

void message_callback(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message) {
    printf("Message arrived on topic: %s\n", message->topic);
    printf("Message: %.*s\n", message->payloadlen, (char *)message->payload);
}

int main() {
    struct mosquitto *mosq = NULL;
    int rc;

    mosquitto_lib_init();
    mosq = mosquitto_new(NULL, true, NULL);
    if (!mosq) {
        fprintf(stderr, "Failed to create client\n");
        mosquitto_lib_cleanup();
        return 1;
    }

    // 设置回调
    mosquitto_message_callback_set(mosq, message_callback);

    rc = mosquitto_connect(mosq, HOST, PORT, 60);
    if (rc != MOSQ_ERR_SUCCESS) {
        fprintf(stderr, "Failed to connect: %s\n", mosquitto_strerror(rc));
        mosquitto_destroy(mosq);
        mosquitto_lib_cleanup();
        return 1;
    }

    // 订阅主题
    rc = mosquitto_subscribe(mosq, NULL, TOPIC, QOS);
    if (rc != MOSQ_ERR_SUCCESS) {
        fprintf(stderr, "Failed to subscribe: %s\n", mosquitto_strerror(rc));
        mosquitto_disconnect(mosq);
        mosquitto_destroy(mosq);
        mosquitto_lib_cleanup();
        return 1;
    }

    printf("Subscribed to topic: %s. Waiting for messages...\n", TOPIC);
    mosquitto_loop_start(mosq); // 启动循环线程
    while(1) {
        sleep(1);
    }

    // 清理（实际中需处理退出信号）
    mosquitto_unsubscribe(mosq, NULL, TOPIC);
    mosquitto_disconnect(mosq);
    mosquitto_destroy(mosq);
    mosquitto_lib_cleanup();
    return 0;
}

编译和运行测试：

  • 编译发布者：
    gcc publisher_mosquitto.c -o publisher_mosquitto -lmosquitto
  • 编译订阅者：
    gcc subscriber_mosquitto.c -o subscriber_mosquitto -lmosquitto
  • 运行测试：
    1. 启动订阅者：./subscriber_mosquitto（后台运行）。
    2. 启动发布者：./publisher_mosquitto。
  • 输出结果：
    • 发布者输出：Message published
    • 订阅者输出：Message arrived on topic: test/topic 和 Message: Hello from Mosquitto Publisher

使用MQTT+TLS实现安全通信

MQTT+TLS通过SSL/TLS加密通信，增加安全性。需要生成证书并配置代理支持TLS。

  1. 生成自签名证书（在Ubuntu上）：
    # 创建CA证书
openssl genrsa -out ca.key 2048
openssl req -new -x509 -days 365 -key ca.key -out ca.crt

# 创建服务器证书
openssl genrsa -out server.key 2048
openssl req -new -key server.key -out server.csr
openssl x509 -req -in server.csr -CA ca.crt -CAkey ca.key -CAcreateserial -out server.crt -days 365

# 配置Mosquitto代理
sudo cp server.crt /etc/mosquitto/certs/
sudo cp server.key /etc/mosquitto/certs/
sudo cp ca.crt /etc/mosquitto/ca_certificates/
sudo nano /etc/mosquitto/conf.d/tls.conf
    在tls.conf中添加：
    listener 8883
cafile /etc/mosquitto/ca_certificates/ca.crt
certfile /etc/mosquitto/certs/server.crt
keyfile /etc/mosquitto/certs/server.key
require_certificate false # 简化测试，不强制客户端证书
    重启代理：sudo systemctl restart mosquitto

  1. 使用paho_mqtt.c实现MQTT+TLS客户端：
    修改发布者和订阅者代码，添加TLS选项。

    发布者MQTT+TLS（publisher_paho_tls.c）：

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

#define ADDRESS "tcp://localhost:1883" // 代理地址
#define CLIENTID "PublisherClient"
#define TOPIC "test/topic"
#define QOS 1
#define TIMEOUT 10000L

int main() {
    MQTTClient client;
    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    MQTTClient_message pubmsg = MQTTClient_message_initializer;
    MQTTClient_deliveryToken token;
    int rc;

    // 创建客户端
    rc = MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    if (rc != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "Failed to create client: %d\n", rc);
        exit(1);
    }

    // 设置连接选项
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
		
	//设置TLS选项
    MQTTClient_SSLOptions ssl_opts = MQTTClient_SSLOptions_initializer;
    conn_opts.ssl = &ssl_opts;
    conn_opts.ssl->trustStore = "/etc/mosquitto/ca_certificates/ca.crt"; // CA证书路径

    // 连接到代理
    rc = MQTTClient_connect(client, &conn_opts);
    if (rc != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "Failed to connect: %d\n", rc);
        MQTTClient_destroy(&client);
        exit(1);
    }

    // 发布消息
    pubmsg.payload = "Hello from Paho Publisher";
    pubmsg.payloadlen = strlen(pubmsg.payload);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    rc = MQTTClient_publishMessage(client, TOPIC, &pubmsg, &token);
    if (rc != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "Failed to publish: %d\n", rc);
        MQTTClient_disconnect(client, TIMEOUT);
        MQTTClient_destroy(&client);
        exit(1);
    }

    printf("Message published. Waiting for delivery token...\n");
    MQTTClient_waitForCompletion(client, token, TIMEOUT);

    // 断开并清理
    rc = MQTTClient_disconnect(client, TIMEOUT);
    if (rc != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "Failed to disconnect: %d\n", rc);
    }
    MQTTClient_destroy(&client);
    return 0;
}

订阅者MQTT+TLS（subscriber_paho_tls.c）：

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

#define ADDRESS "tcp://localhost:1883"
#define CLIENTID "SubscriberClient"
#define TOPIC "test/topic"
#define QOS 1
#define TIMEOUT 10000L

int messageArrived(void *context, char *topicName, int topicLen, MQTTClient_message *message) {
    printf("Message arrived on topic: %s\n", topicName);
    printf("Message: %.*s\n", message->payloadlen, (char *)message->payload);
    MQTTClient_free(topicName);
    MQTTClient_freeMessage(&message);
    return 1;
}

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

    rc = MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    if (rc != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "Failed to create client: %d\n", rc);
        exit(1);
    }

    // 设置回调
    rc = MQTTClient_setCallbacks(client, NULL, NULL, messageArrived, NULL);
    if (rc != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "Failed to set callbacks: %d\n", rc);
        MQTTClient_destroy(&client);
        exit(1);
    }

    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    
	//设置TLS选项
    MQTTClient_SSLOptions ssl_opts = MQTTClient_SSLOptions_initializer;
    conn_opts.ssl = &ssl_opts;
    conn_opts.ssl->trustStore = "/etc/mosquitto/ca_certificates/ca.crt"; // CA证书路径
    
    rc = MQTTClient_connect(client, &conn_opts);
    if (rc != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "Failed to connect: %d\n", rc);
        MQTTClient_destroy(&client);
        exit(1);
    }

    // 订阅主题
    rc = MQTTClient_subscribe(client, TOPIC, QOS);
    if (rc != MQTTCLIENT_SUCCESS) {
        fprintf(stderr, "Failed to subscribe: %d\n", rc);
        MQTTClient_disconnect(client, TIMEOUT);
        MQTTClient_destroy(&client);
        exit(1);
    }

    printf("Subscribed to topic: %s. Waiting for messages...\n", TOPIC);
    while(1) {
        // 保持运行以接收消息
        sleep(1);
    }

    // 清理（实际中需处理退出信号）
    rc = MQTTClient_unsubscribe(client, TOPIC);
    if (rc != MQTTCLIENT_SUCCESS) fprintf(stderr, "Unsubscribe failed: %d\n", rc);
    rc = MQTTClient_disconnect(client, TIMEOUT);
    if (rc != MQTTCLIENT_SUCCESS) fprintf(stderr, "Disconnect failed: %d\n", rc);
    MQTTClient_destroy(&client);
    return 0;
}

编译和运行测试：

  • 编译发布者：
    gcc publisher_paho_tls.c -o publisher_paho_tls -lpaho-mqtt3c -lssl -lcrypto
  • 编译订阅者：
    gcc subscriber_paho_tls.c -o subscriber_paho_tls -lpaho-mqtt3c -lssl -lcrypto
  • 运行测试：
    1. 启动订阅者：./subscriber_paho_tls（后台运行）。
    2. 启动发布者：./publisher_paho_tls。
  • 输出结果：
    • 发布者输出：Message published. Waiting for delivery token...
    • 订阅者输出：Message arrived on topic: test/topic 和 Message: Hello from Paho Publisher

  1. 使用mosquitto库实现MQTT+TLS客户端：
    修改代码添加TLS。

    发布者MQTT+TLS（publisher_mosquitto_tls.c）：

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

#define HOST "localhost"
#define PORT 8883
#define TOPIC "test/topic"
#define QOS 1
#define MESSAGE "Hello from Mosquitto Publisher"

int main() {
    struct mosquitto *mosq = NULL;
    int rc;

    // 初始化库
    mosquitto_lib_init();

    // 创建实例
    mosq = mosquitto_new(NULL, true, NULL);
    if (!mosq) {
        fprintf(stderr, "Failed to create client\n");
        mosquitto_lib_cleanup();
        return 1;
    }

    // 设置TLS
    mosquitto_tls_set(mosq, "/etc/mosquitto/ca_certificates/ca.crt", NULL, NULL, NULL, NULL); // CA证书
    mosquitto_tls_insecure_set(mosq, true); // 允许自签名证书
    
    // 连接到代理
    rc = mosquitto_connect(mosq, HOST, PORT, 60);
    if (rc != MOSQ_ERR_SUCCESS) {
        fprintf(stderr, "Failed to connect: %s\n", mosquitto_strerror(rc));
        mosquitto_destroy(mosq);
        mosquitto_lib_cleanup();
        return 1;
    }

    // 发布消息
    rc = mosquitto_publish(mosq, NULL, TOPIC, strlen(MESSAGE), MESSAGE, QOS, false);
    if (rc != MOSQ_ERR_SUCCESS) {
        fprintf(stderr, "Failed to publish: %s\n", mosquitto_strerror(rc));
        mosquitto_disconnect(mosq);
        mosquitto_destroy(mosq);
        mosquitto_lib_cleanup();
        return 1;
    }

    printf("Message published\n");

    // 断开并清理
    mosquitto_disconnect(mosq);
    mosquitto_destroy(mosq);
    mosquitto_lib_cleanup();
    return 0;
}

编译：gcc publisher_mosquitto_tls.c -o publisher_mosquitto_tls -lmosquitto -lssl -lcrypto

订阅者TLS（subscriber_mosquitto_tls.c）：

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

#define HOST "localhost"
#define PORT 8883
#define TOPIC "test/topic"
#define QOS 1

void message_callback(struct mosquitto *mosq, void *userdata, const struct mosquitto_message *message) {
    printf("Message arrived on topic: %s\n", message->topic);
    printf("Message: %.*s\n", message->payloadlen, (char *)message->payload);
}

int main() {
    struct mosquitto *mosq = NULL;
    int rc;

    mosquitto_lib_init();
    mosq = mosquitto_new(NULL, true, NULL);
    if (!mosq) {
        fprintf(stderr, "Failed to create client\n");
        mosquitto_lib_cleanup();
        return 1;
    }

    // 设置TLS
    mosquitto_tls_set(mosq, "/etc/mosquitto/ca_certificates/ca.crt", NULL, NULL, NULL, NULL); // CA证书
    mosquitto_tls_insecure_set(mosq, true); // 允许自签名证书
    
    // 设置回调
    mosquitto_message_callback_set(mosq, message_callback);

    rc = mosquitto_connect(mosq, HOST, PORT, 60);
    if (rc != MOSQ_ERR_SUCCESS) {
        fprintf(stderr, "Failed to connect: %s\n", mosquitto_strerror(rc));
        mosquitto_destroy(mosq);
        mosquitto_lib_cleanup();
        return 1;
    }

    // 订阅主题
    rc = mosquitto_subscribe(mosq, NULL, TOPIC, QOS);
    if (rc != MOSQ_ERR_SUCCESS) {
        fprintf(stderr, "Failed to subscribe: %s\n", mosquitto_strerror(rc));
        mosquitto_disconnect(mosq);
        mosquitto_destroy(mosq);
        mosquitto_lib_cleanup();
        return 1;
    }

    printf("Subscribed to topic: %s. Waiting for messages...\n", TOPIC);
    mosquitto_loop_start(mosq); // 启动循环线程
    while(1) {
        sleep(1);
    }

    // 清理（实际中需处理退出信号）
    mosquitto_unsubscribe(mosq, NULL, TOPIC);
    mosquitto_disconnect(mosq);
    mosquitto_destroy(mosq);
    mosquitto_lib_cleanup();
    return 0;
}

编译和运行测试：

  • 编译发布者：
    gcc publisher_mosquitto_tls.c -o publisher_mosquitto_tls -lmosquitto
  • 编译订阅者：
    gcc subscriber_mosquitto_tls.c -o subscriber_mosquitto_tls -lmosquitto
  • 运行测试：
    1. 启动订阅者：./subscriber_mosquitto_tls（后台运行）。
    2. 启动发布者：./publisher_mosquitto_tls。
  • 输出结果：
    • 发布者输出：Message published
    • 订阅者输出：Message arrived on topic: test/topic 和 Message: Hello from Mosquitto Publisher

Paho 与 Mosquitto 对比

特性Paho MQTT CMosquitto
定位客户端库代理 + 客户端库
线程模型异步回调同步/异步可选
TLS支持需编译开启默认支持
适用场景嵌入式设备、轻量应用服务端部署、全功能

通过以上步骤，您可以在 Ubuntu 上完整实现 MQTT 通信，并进一步扩展至加密通信场景。

# c语言 # 开发语言 # c++ # 网络 # linux
Logo
乐奇 Rokid 开放社区

更多推荐

在 Rokid AI Glasses 上做背单词应用：AIUI拍照听写的实现

分享在 Rokid AI Glasses 上用 AIUI 开发背单词应用的实践：从镜腿单击拍照、图片编码和多模态识别，到切题时的异步任务处理、语音调度与学习进度保存，结合关键代码和界面动图说明实现过程。

avatar 乐奇 Rokid 开放社区
cover

极刻觅镜 | AI眼镜周报｜智能眼镜周报：销售高增长、标准落地与新品密集发布

avatar 乐奇 Rokid 开放社区
cover

深圳十大 AI 智能眼镜 FPC 厂家盘点（2026 供应商选型参考）

avatar 乐奇 Rokid 开放社区