MQTT Server - 508364/NetLeaf GitHub Wiki

NetLeaf MQTT 服务端模块

概述

netleaf_mqtt_server 是 NetLeaf 的 MQTT v5.0 协议服务端(Broker)模块,完整实现了 MQTT 3.1.1 和 MQTT v5.0 协议规范。支持多客户端连接、主题订阅管理、消息广播等功能。

架构说明

MQTT 服务端作为独立模块,可与客户端模块协同工作:

netleaf_mqtt_server.dll
├── netleaf_mqtt_server.c  (服务端核心功能)
└── netleaf_mqtt_server_lang.h  (多语言支持)

支持的协议版本

  • MQTT 3.1.1
  • MQTT v5.0(完整协议支持)

API 概览

服务器管理

// 创建 MQTT 服务端
nl_mqtt_server_t* nl_mqtt_server_create(void);

// 销毁 MQTT 服务端
void nl_mqtt_server_destroy(nl_mqtt_server_t* server);

// 启动服务端
int nl_mqtt_server_start(nl_mqtt_server_t* server,
                          const nl_mqtt_server_config_t* config);

// 停止服务端
int nl_mqtt_server_stop(nl_mqtt_server_t* server);

// 检查服务端是否运行
int nl_mqtt_server_is_running(const nl_mqtt_server_t* server);

事件回调

// 设置事件回调
int nl_mqtt_server_set_event_callback(
    nl_mqtt_server_t* server,
    nl_mqtt_server_event_callback_t callback,
    void* user_data);

客户端管理

// 获取连接客户端数量
int nl_mqtt_server_get_client_count(const nl_mqtt_server_t* server);

// 断开指定客户端
int nl_mqtt_server_disconnect_client(nl_mqtt_server_t* server,
                                      uint32_t client_id);

// 服务端发布消息
int nl_mqtt_server_publish(nl_mqtt_server_t* server,
                            const char* topic,
                            const void* payload,
                            size_t payload_len,
                            int qos,
                            int retain);

订阅管理

// 遍历订阅
typedef int (*nl_mqtt_server_foreach_subscriber_t)(uint32_t client_id,
                                                    const char* topic,
                                                    int qos,
                                                    void* user_data);

int nl_mqtt_server_foreach_subscription(
    nl_mqtt_server_t* server,
    const char* topic_pattern,
    nl_mqtt_server_foreach_subscriber_t callback,
    void* user_data);

// 获取主题订阅者
int nl_mqtt_server_get_topic_subscribers(
    nl_mqtt_server_t* server,
    const char* topic,
    uint32_t* client_ids,
    size_t* count,
    size_t max_count);

统计数据

typedef struct nl_mqtt_server_stats {
    uint32_t total_connections;
    uint32_t total_disconnections;
    uint32_t total_publishes;
    uint32_t total_subscriptions;
    uint32_t total_unsubscriptions;
    uint32_t bytes_received;
    uint32_t bytes_sent;
} nl_mqtt_server_stats_t;

int nl_mqtt_server_get_stats(const nl_mqtt_server_t* server,
                              nl_mqtt_server_stats_t* stats);

配置选项

typedef struct nl_mqtt_server_config {
    uint16_t                  port;                    // 监听端口
    int                       max_clients;             // 最大客户端数
    int                       tls_enabled;             // 启用 TLS
    int                       auth_enabled;            // 启用认证
    const char*               ca_file;                 // CA 证书
    const char*               cert_file;               // 服务端证书
    const char*               key_file;                // 服务端密钥
    nl_mqtt_server_auth_callback_t auth_callback;      // 认证回调
    void*                     auth_user_data;          // 认证回调数据
    nl_mqtt_property_t*       properties;              // MQTT 5.0 属性
} nl_mqtt_server_config_t;

事件类型

typedef enum {
    NL_MQTT_SERVER_EVT_CONNECT    = 0,   // 客户端连接
    NL_MQTT_SERVER_EVT_DISCONNECT = 1,   // 客户端断开
    NL_MQTT_SERVER_EVT_SUBSCRIBE  = 2,   // 客户端订阅
    NL_MQTT_SERVER_EVT_UNSUBSCRIBE= 3,   // 客户端取消订阅
    NL_MQTT_SERVER_EVT_PUBLISH    = 4    // 收到发布消息
} nl_mqtt_server_event_type_t;

使用示例

基本服务端

#include "netleaf_mqtt_server.h"
#include <stdio.h>

void on_event(const nl_mqtt_server_event_t* event, void* userdata) {
    switch (event->type) {
        case NL_MQTT_SERVER_EVT_CONNECT:
            printf("Client %u connected\n", event->client_id);
            break;
        case NL_MQTT_SERVER_EVT_DISCONNECT:
            printf("Client %u disconnected\n", event->client_id);
            break;
        case NL_MQTT_SERVER_EVT_PUBLISH:
            printf("Message on '%s': %.20s...\n",
                   event->topic, (const char*)event->payload);
            break;
    }
}

int main() {
    nl_mqtt_server_t* server = nl_mqtt_server_create();
    if (!server) {
        printf("Failed to create server\n");
        return 1;
    }

    // 设置事件回调
    nl_mqtt_server_set_event_callback(server, on_event, NULL);

    // 配置服务端
    nl_mqtt_server_config_t config = {0};
    config.port = 1883;
    config.max_clients = 100;

    // 启动服务端
    int ret = nl_mqtt_server_start(server, &config);
    if (ret != 0) {
        printf("Failed to start server: %d\n", ret);
        nl_mqtt_server_destroy(server);
        return 1;
    }

    printf("MQTT Server running on port %d\n", config.port);
    printf("Max clients: %d\n", config.max_clients);

    // 保持运行(实际应用中应处理事件循环)
    while (nl_mqtt_server_is_running(server)) {
        // 处理客户端连接
        // 注意:这是一个简化的示例,实际实现需要异步处理
    }

    // 清理
    nl_mqtt_server_stop(server);
    nl_mqtt_server_destroy(server);
    return 0;
}

带认证的服务器

#include "netleaf_mqtt_server.h"
#include <stdio.h>
#include <string.h>

// 简单的认证回调
int my_auth_callback(const char* client_id,
                      const char* username,
                      const char* password,
                      void* user_data) {
    // 这里应该查询数据库或配置文件验证
    if (strcmp(username, "admin") == 0 &&
        strcmp(password, "secret") == 0) {
        return 0;  // 认证成功
    }
    return -1;  // 认证失败
}

int main() {
    nl_mqtt_server_t* server = nl_mqtt_server_create();

    nl_mqtt_server_config_t config = {0};
    config.port = 1883;
    config.max_clients = 50;
    config.auth_enabled = 1;
    config.auth_callback = my_auth_callback;

    nl_mqtt_server_start(server, &config);

    // ... 运行服务端

    nl_mqtt_server_stop(server);
    nl_mqtt_server_destroy(server);
    return 0;
}

服务端发布消息

#include "netleaf_mqtt_server.h"
#include <stdio.h>

int main() {
    nl_mqtt_server_t* server = nl_mqtt_server_create();

    // 启动服务端
    nl_mqtt_server_config_t config = {0};
    config.port = 1883;
    nl_mqtt_server_start(server, &config);

    // 发布消息到指定主题
    const char* msg = "Temperature: 25.5°C";
    nl_mqtt_server_publish(server, "sensors/temp", msg, strlen(msg), 1, 0);

    // 获取统计信息
    nl_mqtt_server_stats_t stats = {0};
    nl_mqtt_server_get_stats(server, &stats);
    printf("Total connections: %u\n", stats.total_connections);
    printf("Total publishes: %u\n", stats.total_publishes);

    nl_mqtt_server_stop(server);
    nl_mqtt_server_destroy(server);
    return 0;
}

多语言错误消息

#include "netleaf_mqtt_server.h"
#include "netleaf_lang.h"
#include <stdio.h>

int main() {
    nl_mqtt_server_init();

    // 设置中文
    nl_lang_set("zh_cn");

    nl_mqtt_server_t* server = nl_mqtt_server_create();
    nl_mqtt_server_config_t config = {0};
    config.port = 99999;  // 无效端口

    int ret = nl_mqtt_server_start(server, &config);
    if (ret != 0) {
        const char* msg = nl_lang_get_error(NL_LIB_MQTT_SERVER, ret);
        printf("错误: %s (原始码: %d)\n", msg, ret);
        // 中文输出: 错误: Socket绑定失败 (原始码: -2)
    }

    nl_mqtt_server_destroy(server);
    return 0;
}

构建配置

CMake 选项

# 仅构建 MQTT 服务端
cmake -B build -DBUILD_MQTT_SERVER=ON

# 构建 MQTT 客户端 + 服务端
cmake -B build -DBUILD_MQTT=ON -DBUILD_MQTT_SERVER=ON

# 构建全部(含 TLS)
cmake -B build -DBUILD_MQTT=ON -DBUILD_MQTT_SERVER=ON -DBUILD_TLS=ON

构建脚本

# Windows
build_all-MSVC.bat
build_all-Clang.bat

# Linux/macOS
./build_all-linux.sh
./build_all-macos.sh

协议特性

支持的消息类型

  • CONNECT / CONNACK
  • PUBLISH / PUBACK / PUBREC / PUBREL / PUBCOMP
  • SUBSCRIBE / SUBACK
  • UNSUBSCRIBE / UNSUBACK
  • PINGREQ / PINGRESP
  • DISCONNECT

QoS 等级

QoS 说明
0 最多一次(fire and forget)
1 至少一次(带确认)
2 恰好一次(四次握手)

主题通配符

  • + - 匹配单个路径段
  • # - 匹配剩余所有路径段

示例:

  • sensor/temp/+ - 匹配 sensor/temp/live、sensor/temp/history
  • sensor/# - 匹配 sensor/temp、sensor/temp/live、sensor/humidity

MQTT 5.0 特性

  • 会话过期间隔
  • 主题别名
  • 最大包大小
  • 请求问题信息
  • 用户属性

开源项目来源

本模块参考了以下开源项目:

项目 用途 许可证 仓库链接
MQTT Specification MQTT v5.0 协议规范 EPL-2.0 / EDL 1.0 mqtt.org
Mosquitto MQTT Broker 参考实现 EDL 1.0 / CPL 1.0 eclipse/mosquitto
Paho MQTT C MQTT C 客户端参考实现 EPL-2.0 / EDL 1.0 eclipse/paho.mqtt.c

注意事项

  1. 端口绑定:默认端口 1883,如需更改请修改配置
  2. 客户端限制:默认最大 1024 个并发客户端
  3. 主题验证:服务端会自动验证主题格式
  4. 线程安全:服务端核心非线程安全,多线程使用需加锁
  5. 内存管理:消息回调中不要保留 payload 指针

相关文档

⚠️ **GitHub.com Fallback** ⚠️