MQTT Server

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 指针

相关文档

• API 参考 - 完整 API 列表
• MQTT 客户端 - 客户端模块文档
• TLS 模块 - TLS/SSL 加密支持
• 功能特性 - MQTT 功能介绍
• 示例代码 - 使用示例