使用手册

快速开始

MQHub 是单二进制文件，无需安装任何运行时，下载即可运行。

1. 赋权并启动

# 赋予可执行权限
chmod +x mqhub

# 前台启动（默认监听 TCP 1883、Web 8081）
./mqhub

2. 打开管理面板

浏览器访问 http://服务器IP:8081，默认账号密码均为 admin。首次登录后请立即修改 Web 管理员密码。

3. 首次连接 MQTT

首次启动会自动创建一个 MQTT 用户（默认 admin / change-me），可通过环境变量 MQTT_ADMIN_USER 和 MQTT_ADMIN_PASS 修改。

# 订阅测试
mosquitto_sub -h 127.0.0.1 -p 1883 -u admin -P change-me -t '#' -v

# 发布测试
mosquitto_pub -h 127.0.0.1 -p 1883 -u admin -P change-me -t test -m "hello"

默认端口

MQHub 默认开启 TCP、WebSocket 两个监听；TLS / WSS / QUIC 需手动配置证书后启用。

环境变量

MQHub 所有配置均通过环境变量注入，无需配置文件。

基础配置

存储与集群

性能调优

TLS / QUIC 证书

GreptimeDB 时序落库

MySQL 落库

PostgreSQL 落库

存储后端

MQHub 支持 SQLite（单机零依赖）和 Redis（集中 / 集群共享）两种后端。

SQLite（默认）

零外部依赖，数据库文件默认位于 ./data/mqhub.db。启用 WAL 模式与 PRAGMA 调优后，读写不互斥。适合单机、测试环境。

STORAGE_BACKEND=sqlite \
MQTT_DB_PATH=/var/lib/mqhub/mqhub.db \
SQLITE_CACHE_SIZE=-256000 \
SQLITE_MMAP_SIZE=1073741824 \
./mqhub

Redis

用户与 ACL 存储在 Redis，多节点可共享。集群模式必须使用 Redis。使用共享连接池，与事件总线共用一个池。

STORAGE_BACKEND=redis \
REDIS_URL=redis://127.0.0.1:6379 \
REDIS_POOL_MAX=128 \
./mqhub

单机对比

TLS 加密

配置证书后，MQHub 会在 8883（MQTT TLS）和 8443（WSS）两个端口同时启用 TLS，共用同一份证书。

MQTT_TLS_CERT=/etc/mqhub/server.crt \
MQTT_TLS_KEY=/etc/mqhub/server.key \
MQTT_TLS_PORT=8883 \
./mqhub

证书要求

• 格式：PEM 编码
• 证书与私钥必须匹配
• 如使用自签证书，客户端需信任该 CA

QUIC 传输

QUIC 是基于 UDP 的新一代传输协议，提供 0-RTT 连接、无队头阻塞、连接迁移，弱网环境下表现优于 TCP。

前置条件

• 证书：配置 MQTT_TLS_CERT / MQTT_TLS_KEY（与 TLS 复用同一份）
• 防火墙：开放 UDP 端口（如 ufw allow 9443/udp）
• 客户端：必须支持 QUIC（paho / esp-mqtt / mosquitto 均不支持）

启用示例

MQTT_TLS_CERT=./certs/server.pem \
MQTT_TLS_KEY=./certs/server.key \
MQTT_QUIC_PORT=9443 \
./mqhub

启动日志

[main] TLS listener enabled on port 8883
[main] WSS listener enabled on port 8443
[main] QUIC listener enabled on port 9443
MQTT Broker Listening on quic 0.0.0.0:9443

验证

# 确认 UDP 端口监听
ss -ulnp | grep 9443

集群模式

通过 CLUSTER_MODE 环境变量切换部署模式，集群模式必须使用 Redis 存储后端。

standalone（单机）

默认模式，无集群协调，单节点全功能。

CLUSTER_MODE=standalone \
./mqhub

raft（强一致集群）

基于 Raft 协议，强一致性，防止 Client ID 重复登录。生产环境首选。建议 3 节点起步（容忍 1 节点故障）。

CLUSTER_MODE=raft \
STORAGE_BACKEND=redis \
REDIS_URL=redis://127.0.0.1:6379 \
REDIS_POOL_MAX=64 \
CLUSTER_NODE_ID=node-1 \
CLUSTER_RAFT_CONFIG=./mqhub-cluster-raft.toml \
./mqhub

broadcast（高吞吐集群）

无 Leader，去中心化，gRPC 广播同步。追求极致吞吐与低延迟。

CLUSTER_MODE=broadcast \
STORAGE_BACKEND=redis \
REDIS_URL=redis://127.0.0.1:6379 \
REDIS_POOL_MAX=128 \
CLUSTER_NODE_ID=node-1 \
CLUSTER_BROADCAST_CONFIG=./mqhub-cluster-broadcast.toml \
./mqhub

集群特性

集群模式下，MQHub 通过 Redis Pub/Sub 事件总线实现跨节点协调：

• ACL 缓存即时失效：任一节点修改用户后，通过 Redis Pub/Sub 广播失效事件，所有节点在毫秒级（<100ms）同步失效本地缓存
• 全局在线客户端视图：Web 面板可查看全集群在线客户端，通过 is_remote 字段区分本地与远程节点
• 缓存清空广播：Web 面板点击「清空缓存」会广播到所有节点
• 节点心跳发现：每个节点每 10 秒广播一次心跳，30 秒无心跳自动从节点列表移除
• Redis 连接池共享：storage 与 cluster_bus 共用一个连接池
• 落库天然去重：GreptimeDB / MySQL / PostgreSQL 三个落库写入器都使用共享订阅，每条消息只投递给每组的一个节点，无需 Redis 分布式锁

节点视图

Web 面板的「集群节点」卡片显示当前存活节点数，点击可查看每个节点的详细信息（主机名、PID、版本、运行时长、最后心跳）。

心跳机制说明：

MQTT 桥接

MQTT 桥接用于连接两个独立的 MQHub 实例，实现跨网络、跨机房、边缘到云端的数据同步。与集群模式不同，桥接不共享路由表，仅按主题转发消息。

两种方向

启用方式

桥接为可选功能，通过配置文件的存在与否控制。配置文件放在程序工作目录下即可。

Egress 配置示例

# ./mqhub-bridge-egress.toml
[[bridges]]
enable = true
name = "to_center"
client_id_prefix = "mqhub-edge-up"
server = "center-mqhub.example.com:1883"
username = "edge_bridge"
password = "your-password"
concurrent_client_limit = 3
connect_timeout = "20s"
keepalive = "60s"
reconnect_interval = "5s"
message_channel_capacity = 100000
mqtt_ver = "v5"

[[bridges.entries]]
local.topic_filter = "factory/#"
remote.topic = "central/factory/#"

Ingress 配置示例

# ./mqhub-bridge-ingress.toml
[[bridges]]
enable = true
name = "from_center"
client_id_prefix = "mqhub-edge-down"
server = "center-mqhub.example.com:1883"
username = "edge_bridge"
password = "your-password"
concurrent_client_limit = 3

[[bridges.entries]]
remote.topic_filter = "cmd/#"
local.topic = "local/cmd/#"
local.qos = 1
local.retain = false

共享订阅

共享订阅让同一组内的多个订阅者共同消费消息——每条消息只投递给组内一个订阅者。

启用方式

放置 mqhub-shared-subscription.toml 即启用：

# ./mqhub-shared-subscription.toml
strategy = "hash_topic"
sticky_cache_size = 100000

订阅格式

$share/{Group}/{TopicFilter}

示例

# 两个订阅者加入同一组 workers
mosquitto_sub -t '$share/workers/sensor/+/temp' -v

# 发布 10 条消息，两个订阅者各收到约 5 条
mosquitto_pub -t 'sensor/room1/temp' -m '25'

策略

插件配置文件

部分功能通过独立的 TOML 配置文件启用，采用"文件存在即启用，缺失则跳过"策略。

所有配置文件放在程序工作目录下，重启即生效：

会话持久化（session-storage）

启用后 Broker 重启不再丢失客户端订阅关系和离线消息，客户端重连后订阅自动恢复。

与内存模式对比

前置条件

需要在 Cargo.toml 启用对应的 feature：

rmqtt-session-storage = { version = "0.24", features = ["sled"] }
# 或 features = ["redis"] / ["sled", "redis", "redis-cluster"]

配置示例（Sled 后端，单机）

# ./mqhub-session-storage.toml
storage.type = "sled"
storage.sled.path = "/var/log/mqhub/.cache/session/{node}"
storage.sled.cache_capacity = "3G"

配置示例（Redis 后端，集群）

# ./mqhub-session-storage.toml
storage.type = "redis"
storage.redis.url = "redis://127.0.0.1:6379/"
storage.redis.prefix = "session-{node}"

消息持久化（message-storage）

启用后所有未过期的消息会被持久化存储，新订阅者能收到历史消息，Broker 重启后消息不丢。

与 Retain 消息的区别

适用场景

• 设备状态同步：新上线的监控客户端立即获取设备最新状态
• 配置分发：新接入的网关立刻获取已有配置指令
• 数据回溯：分析服务重启后拉取断连期间的全部数据

前置条件

rmqtt-message-storage = { version = "0.24", features = ["ram", "redis"] }

配置示例（RAM 后端）

# ./mqhub-message-storage.toml
storage.type = "ram"
storage.ram.cache_capacity = "3G"
storage.ram.cache_max_count = 1000000
storage.ram.encode = false
cleanup_count = 5000

配置示例（Redis 后端）

# ./mqhub-message-storage.toml
storage.type = "redis"
storage.redis.url = "redis://127.0.0.1:6379/"
storage.redis.prefix = "message-{node}"
cleanup_count = 5000

GreptimeDB 落库

订阅指定 MQTT 主题，自动将消息批量写入 GreptimeDB 时序数据库。

采用共享订阅（$share/greptime_writer/...）：Broker 保证每条消息只投递给组内一个节点，天然去重，无需 Redis 分布式锁，也不依赖 Redis。

表结构

MQHub 会自动创建目标表（若不存在）：

启用示例

GREPTIMEDB_ENABLED=1 \
GREPTIMEDB_ADDR=127.0.0.1:4001 \
GREPTIMEDB_DB=public \
GREPTIMEDB_TABLE=sensor_readings \
GREPTIMEDB_TOPICS=sensors/#,devices/+/telemetry \
GREPTIMEDB_SHARE_GROUP=greptime_writer \
GREPTIMEDB_BATCH_SIZE=2000 \
./mqhub

MySQL 落库

订阅指定 MQTT 主题，自动将消息批量写入 MySQL 数据库。与 GreptimeDB 落库原理相同，只是目标数据库不同。

表结构

MQHub 会自动创建目标表（若不存在）：

CREATE TABLE IF NOT EXISTS `sensor_readings` (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    topic VARCHAR(255) NOT NULL,
    device_id VARCHAR(255) NOT NULL,
    ts BIGINT NOT NULL,
    payload LONGTEXT,
    payload_size BIGINT NOT NULL,
    INDEX idx_topic (topic),
    INDEX idx_device_id (device_id),
    INDEX idx_ts (ts)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4

启用示例

MYSQL_ENABLED=1 \
MYSQL_URL=mysql://root:password@127.0.0.1:3306/mqhub \
MYSQL_TABLE=sensor_readings \
MYSQL_TOPICS=sensors/#,devices/+/telemetry \
MYSQL_SHARE_GROUP=mysql_writer \
MYSQL_BATCH_SIZE=2000 \
MYSQL_FLUSH_INTERVAL_MS=1000 \
./mqhub

集群模式

集群模式下，采用共享订阅（$share/mysql_writer/...），Broker 保证每条消息只投递给组内一个节点，天然去重，无需 Redis 锁。

内容转换（可选）

开启后，MQTT 的 payload 会被解析成结构化字段，和原始 payload 一起写入数据库。默认关闭。

启用示例

MYSQL_TRANSFORM_ENABLED=1 \
                    MYSQL_TRANSFORM_RULE=json_extract \
                    MYSQL_TRANSFORM_FIELDS='{"temp":"$.temperature","hum":"$.humidity"}' \
                    ./mqhub

存储结构对比

不开启转换（默认）时，payload 原样存进 payload 列：

开启转换后，字段被拆开，同时原始 payload 保留：

转换规则

失败策略

ALTER TABLE sensor_readings
                        ADD COLUMN temp DOUBLE,
                        ADD COLUMN hum BIGINT;

PostgreSQL 落库

订阅指定 MQTT 主题，自动将消息批量写入 PostgreSQL 数据库。

表结构

MQHub 会自动创建目标表（若不存在）：

CREATE TABLE IF NOT EXISTS "sensor_readings" (
    id BIGSERIAL PRIMARY KEY,
    topic VARCHAR(255) NOT NULL,
    device_id VARCHAR(255) NOT NULL,
    ts BIGINT NOT NULL,
    payload TEXT,
    payload_size BIGINT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_sensor_readings_topic ON "sensor_readings"(topic);
CREATE INDEX IF NOT EXISTS idx_sensor_readings_device_id ON "sensor_readings"(device_id);
CREATE INDEX IF NOT EXISTS idx_sensor_readings_ts ON "sensor_readings"(ts);

启用示例

PSQL_ENABLED=1 \
PSQL_URL=postgres://postgres:password@127.0.0.1:5432/mqhub \
PSQL_TABLE=sensor_readings \
PSQL_TOPICS=sensors/#,devices/+/telemetry \
PSQL_SHARE_GROUP=psql_writer \
PSQL_BATCH_SIZE=2000 \
PSQL_FLUSH_INTERVAL_MS=1000 \
./mqhub

集群模式

与 MySQL 落库相同，采用共享订阅（$share/psql_writer/...），Broker 保证每条消息只投递给组内一个节点，天然去重，无需 Redis 锁。

内容转换（可选）

与 MySQL 相同，把 payload 解析成结构化字段。配置项前缀为 PSQL_。

PSQL_TRANSFORM_ENABLED=1 \
                    PSQL_TRANSFORM_RULE=json_extract \
                    PSQL_TRANSFORM_FIELDS='{"temp":"$.temperature","hum":"$.humidity"}' \
                    ./mqhub

Web 管理面板

访问 http://服务器IP:8081，使用 Basic Auth 登录（默认 admin / admin）。

功能一览

• 实时数据：8 个实时指标卡片 + 热门主题 TOP 10 + 协议分布
• 用户管理：增删改查 MQTT 用户，编辑 ACL 规则
• 在线客户端：实时查看所有连接中的客户端，支持踢下线
• 操作审计日志：所有管理操作的审计记录，含操作者 IP
• 消息追踪：实时追踪匹配主题的消息事件
• 实时统计：点击顶部「统计」按钮，查看完整指标快照（含消息吞吐量）
• 集群节点：点击「集群节点」卡片，查看当前存活的所有节点详情
• MQTT 测试工具：内置客户端，支持连接、发布、订阅、查看消息日志
• 分页与搜索：所有列表支持分页与关键字搜索
• 全局自动刷新：header 下拉选择刷新间隔（不刷新 / 3s / 5s / 10s / 30s / 1min）
• 修改密码：修改 Web 管理员密码

顶部按钮

集群模式下的视图

• header 显示的「在线」数量为全集群总数
• 列表中标有「远程」标记的为其他节点的客户端
• 远程客户端不显示「踢下线」按钮，也无法查看订阅详情

用户管理

每个 MQTT 用户包含：用户名、密码、启用状态、超级用户标记、ACL 规则列表。

字段说明

ACL 规则

每条 ACL 规则一行，格式为 access action clientid topics，字段之间用空格分隔。

字段说明

匹配规则

• 多条规则按顺序匹配，第一条匹配的规则决定结果
• deny 规则建议放在前面
• 若没有任何规则命中，默认拒绝（白名单模式）

示例

# 设备只能发布到自己的主题、订阅自己的控制主题
allow publish   device-* devices/device-*/telemetry
allow subscribe device-* devices/device-*/cmd

# 管理后台可以访问所有主题
allow pubsub    admin-*  #

# 所有客户端禁止订阅 $SYS（deny 放前面优先命中）
deny  subscribe *        $SYS/#

# 允许所有客户端订阅公共广播
allow subscribe *        public/#

保留消息（Retain）

保留消息是 MQTT 协议的标配功能：发布时设置 retain 标志，Broker 会为每个主题保存最新一条消息。新订阅者订阅该主题时，会立即收到这条保留消息。

启用方式

保留消息为可选功能，需同时满足两个条件才启用：

1. 编译时：Cargo.toml 里包含 rmqtt-retainer 依赖
2. 运行时：程序工作目录下存在 mqhub-retainer.toml 配置文件

配置文件

storage.type = "ram"
max_retained_messages = 10000
max_payload_size = "256KB"
retained_message_ttl = "24h"

三种后端对比

匿名接入

MQHub 默认采用「用户名 + 密码 + ACL」的白名单模式。可通过环境变量开启匿名接入。

开启方式

MQTT_ALLOW_ANONYMOUS=1 \
./mqhub

开启后行为

• 未提供用户名的客户端直接通过认证
• 匿名客户端不经过任何 ACL 检查
• 已提供用户名的客户端仍走正常流程

客户端接入

MQHub 兼容所有标准 MQTT 客户端。以下为常用示例。

命令行测试（mosquitto）

# 订阅
mosquitto_sub -h 127.0.0.1 -p 1883 \
  -u device-001 -P your-password \
  -t 'devices/device-001/cmd' -v

# 发布 QoS 1
mosquitto_pub -h 127.0.0.1 -p 1883 \
  -u device-001 -P your-password \
  -t 'devices/device-001/telemetry' \
  -q 1 -m '{"temp":25.3}'

WebSocket 接入

浏览器 / 小程序通过 ws://服务器IP:8080/mqtt 接入。

// MQTT.js 示例
import mqtt from 'mqtt'

const client = mqtt.connect('ws://your-server:8080/mqtt', {
  username: 'device-001',
  password: 'your-password',
  clientId: 'web-client-001'
})

client.on('connect', () => {
  client.subscribe('devices/+/telemetry')
})

TLS 接入

mosquitto_sub -h your-server -p 8883 \
  --cafile /path/to/ca.crt \
  -u device-001 -P your-password \
  -t '#' -v

QUIC 接入

QUIC 走 UDP，端口 9443。仅少数客户端（如 Rust mqtt5）支持。

// Rust mqtt5 客户端（示例）
let client = mqtt5::Client::connect("quic://your-server:9443").await?;

许可激活

MQHub 首次启动时会检测 license.lic 文件。若文件不存在或无效，程序会将机器码打印到 stderr。

激活流程

1. 运行 MQHub，在控制台找到形如「请激活，机器码: XXXXXXXX...」的输出
2. 将机器码发送给授权方，获取序列号
3. 将序列号（Base32 编码）写入程序同目录的 license.lic 文件
4. 重新启动 MQHub

# 写入序列号（示例）
echo 'ABCDEFGH...' > license.lic

# 重新启动
./mqhub

日志与调试

通过 RUST_LOG 环境变量控制日志级别。生产环境推荐 info，排障时用 debug。

# 只输出警告和错误
RUST_LOG=warn ./mqhub

# 详细日志（含每次连接、ACL 检查）
RUST_LOG=debug ./mqhub

# 只针对 mqhub-auth 插件开启 debug
RUST_LOG=info,mqhub_auth=debug ./mqhub

日志前缀

客户端踢下线

当出现 Client ID 冲突、异常连接或权限变更后需要强制旧连接重连时，可以踢客户端下线。

使用场景

• Client ID 冲突：同一 Client ID 重复登录，需要踢掉旧连接
• 异常客户端：行为异常或疑似被攻击的客户端需要强制断开
• 权限变更：ACL 变更后，希望旧连接重新认证
• 强制运维：需要立即让某设备离线

操作方式

Web 面板 → 在线客户端列表 → 找到目标客户端 → 点击「踢下线」按钮。

1. 打开 Web 面板，进入「在线客户端」卡片
2. 搜索或翻页找到目标客户端
3. 点击该行「踢下线」按钮
4. 确认后客户端立即断开

原理

通过 RMQTT 的 Entry::kick(clean_start, clear_subscriptions, is_admin) 实现：

审计记录

每次踢下线操作会写入审计日志：

{
  "ts": 1735689600000,
  "operator": "admin",
  "action": "kick_client",
  "target": "device-001",
  "detail": "{\"clean_start\":true,\"clear_subscriptions\":false}",
  "ip": "192.168.1.100",
  "result": "success"
}

消息追踪

在 MQTT publish 路径上实时记录消息事件，用于排障。追踪是临时工具，不是长期监控。

使用场景

• 设备说"发了消息"但服务端没收到：追踪能看出消息是否到达 Broker
• 消息到了但被 ACL 拒绝：追踪直接显示 DENY，指明具体哪条消息被拦
• 验证 ACL 规则改动是否生效：改完规则后发一条测试消息，看追踪结果
• 排查消息丢失：对比"事件数"和"订阅者收到数"，定位问题环节
• 追踪特定设备的消息流：通过 client_id 过滤，只看目标设备

操作方式

Web 面板 → header「追踪」按钮 → 填写参数 → 开始。

参数说明

事件显示

追踪开启后，弹窗底部每 1.5 秒刷新一次，显示匹配的消息事件：

12:34:56.789  PASS  sensors/temp   device-001  25B QoS1
12:34:57.123  PASS  sensors/hum    device-002  18B QoS0
12:34:57.456  DENY  sensors/secret device-003  10B QoS1 · R

性能影响

操作审计日志

记录所有管理操作（用户管理、清空缓存、踢客户端、消息追踪等），用于追溯与合规审计。

记录内容

操作类型

IP 字段说明

审计日志的 ip 记录的是访问 Web 面板的人的 IP（操作者），不是 MQTT 设备的 IP。

存储上限

使用方式

Web 面板 → 底部「操作审计日志」卡片：

• 支持按操作类型过滤（下拉选择）
• 支持分页（10 / 20 / 50 / 100 条每页）
• 点 IP 可一键复制

实时统计

点击顶部「统计」按钮打开实时统计弹窗，查看 Broker 的完整指标快照（含消息吞吐量）。

按需启用机制

为避免日常运行产生额外开销，消息吞吐量统计采用「读取即启用、超时自动关闭」机制：

• 打开弹窗自动启用：每次读取统计快照，会打开 PUBLISH 消息计数开关并续期 30 秒
• 无访问自动关闭：30 秒内没有新的读取，后台 watchdog 自动关闭计数开关
• 关闭时零开销：开关关闭时，publish 路径只做一次原子读（~1ns），不产生任何写操作
• 启用时精确统计：开关打开时，每条 PUBLISH 消息增加一次原子计数（~3ns）

显示内容

弹窗按四组展示指标：

消息流量

连接会话

订阅路由

队列与集群

统计口径

• 累计消息 只统计普通用户的 PUBLISH。超级用户绕过 ACL hook，不会被计数
• 其余字段来自 RMQTT 内置的 scx.stats 快照
• 握手累计是累计值，不会随时间下降，除非重启 Broker

性能影响

系统调优

MQHub 在高并发场景下需要对操作系统、应用参数与 systemd 做相应调整。

1. 文件描述符限制

每个 TCP 连接消耗一个文件描述符（FD）。系统默认限制通常为 1024，远不足以支撑实际负载。

systemd 服务配置（推荐）

[Unit]
Description=MQHub MQTT Broker
After=network.target

[Service]
Type=simple
User=mqhub
Group=mqhub
WorkingDirectory=/opt/mqhub
ExecStart=/opt/mqhub/mqhub

# 环境变量
Environment="MQTT_TCP_PORT=1883"
Environment="WEB_PORT=8081"
Environment="STORAGE_BACKEND=sqlite"
Environment="MQTT_DB_PATH=/var/lib/mqhub/mqhub.db"
Environment="RUST_LOG=info"

# 性能调优（4GB 边缘机器）
Environment="ACL_CACHE_MAX=50000"
Environment="WEB_PER_PAGE_MAX=50"
Environment="SQLITE_CACHE_SIZE=-64000"
Environment="SQLITE_MMAP_SIZE=268435456"
Environment="MIMALLOC_PURGE_DELAY=0"
Environment="MIMALLOC_ARENA_EAGER_COMMIT=0"

# 文件描述符 / 进程限制
LimitNOFILE=1048576
LimitNPROC=1048576

Restart=on-failure
RestartSec=5s

[Install]
WantedBy=multi-user.target

2. 内核参数优化

编辑 /etc/sysctl.conf，修改后执行 sudo sysctl -p 生效。

# ===== 文件描述符 =====
fs.file-max = 2097152

# ===== 连接队列（应对连接风暴）=====
net.core.somaxconn = 32768
net.ipv4.tcp_max_syn_backlog = 65536

# ===== TIME_WAIT 回收 =====
net.ipv4.tcp_tw_reuse = 1
net.ipv4.tcp_fin_timeout = 30

# ===== TCP Keepalive（探测死连接）=====
net.ipv4.tcp_keepalive_time = 600
net.ipv4.tcp_keepalive_intvl = 30
net.ipv4.tcp_keepalive_probes = 3

# ===== 本地端口范围 =====
net.ipv4.ip_local_port_range = 1024 65535

# ===== 网络缓冲区（提升吞吐）=====
net.core.rmem_max = 134217728
net.core.wmem_max = 134217728
net.ipv4.tcp_rmem = 4096 87380 134217728
net.ipv4.tcp_wmem = 4096 65536 134217728

3. 应用层调优（按硬件规模）

4GB 边缘机器

ACL_CACHE_MAX=50000
WEB_PER_PAGE_MAX=50
SQLITE_CACHE_SIZE=-64000
SQLITE_MMAP_SIZE=268435456
MIMALLOC_PURGE_DELAY=0
MIMALLOC_ARENA_EAGER_COMMIT=0
./mqhub

128GB 生产机器

ACL_CACHE_MAX=1000000
WEB_PER_PAGE_MAX=500
SQLITE_CACHE_SIZE=-256000
SQLITE_MMAP_SIZE=1073741824
MIMALLOC_PURGE_DELAY=1000
MIMALLOC_ARENA_EAGER_COMMIT=1
./mqhub

反向代理场景

# Nginx 需转发真实客户端 IP
TRUST_PROXY=1

4. 验证生效

# 查看进程文件描述符限制
cat /proc/<mqhub-PID>/limits | grep "Max open files"

# 查看内核参数
sysctl net.core.somaxconn net.ipv4.tcp_max_syn_backlog

# 观察连接数
ss -s

常见问题

Q：忘记 Web 管理员密码怎么办？

删除数据库中的 admin_config 表相关记录，然后重启 MQHub，会用 WEB_USER / WEB_PASS 环境变量重新初始化。

# SQLite 后端
sqlite3 ./data/mqhub.db \
  "DELETE FROM admin_config WHERE key IN ('web_user','web_pass')"

Q：修改了用户 ACL 但不生效？

通过 Web 面板修改会自动清缓存，无需额外操作。集群模式下通过 Redis Pub/Sub 广播，所有节点在毫秒级（<100ms）生效。若是直接改了数据库文件，点击 Web 面板右上角「清空缓存」即可。

Q：为什么连接成功但发布 / 订阅被拒绝？

认证通过 ≠ 有权限。检查该用户的 ACL 规则：是否包含对应的 action、clientid 是否匹配、主题是否命中。可以开启消息追踪，看具体哪条消息被 DENY。

Q：端口被占用怎么办？

通过环境变量修改端口，比如 MQTT_TCP_PORT=18830。

Q：集群模式下为什么建议必须用 Redis？

多节点需要共享用户与 ACL 数据，同时集群事件广播（ACL 失效、在线表同步、节点心跳）也依赖 Redis Pub/Sub。SQLite 是本地文件，无法保证配置一致。

Q：集群模式下改了 ACL，为什么节点 B 也立即生效？

MQHub 通过 Redis Pub/Sub 事件总线实现跨节点协调。任一节点修改用户后，会立即在 mqhub:cluster:events 通道发布失效事件，所有节点订阅该通道并在收到事件后清空本地 ACL 缓存，因此生效延迟通常在 <100ms。

Q：集群节点为什么显示 1 个？

如果确实部署了多节点但只显示 1 个，可能是：

• Redis 不可达，节点之间互相看不见（需结合 Redis 监控排查）
• 其他节点启动不到 10 秒，心跳还没广播出去
• 节点的 REDIS_URL 配置不一致，连到了不同的 Redis

点开「集群节点」卡片可以看到每个节点的最后心跳时间，用于判断节点是否活跃。

Q：集群模式下为什么查询远程客户端的订阅会返回 501？

订阅详情和踢下线都需要访问客户端的完整 Session 对象，该对象只存在于客户端实际连接的节点。其他节点只有摘要信息，无法操作。返回 501 提示用户访问正确的节点。

Q：Broker 重启后订阅关系会丢吗？

默认会话存储在内存中，Broker 重启会丢失。有两种方案：

• 改客户端：在 ConnAck 后重新订阅（推荐，零成本）
• 启用 session-storage：将订阅关系和离线消息持久化到 Sled / Redis，Broker 重启后客户端重连自动恢复

注意：session-storage 仅对 clean_session=false 的客户端生效。使用 clean_session=true 的客户端按协议规定无法恢复。

Q：新订阅者能收到历史消息吗？

两种机制：

• retain 消息：只存 retain=true 的最后一条，新订阅者立即收到
• message-storage 插件：存储所有未过期消息，新订阅者收到全部历史消息（过期时间内）

需要"任何历史消息都能回溯"的场景用 message-storage；只需要"最新状态"的场景用 retain 消息即可。

Q：QUIC 该怎么用？

配置 MQTT_TLS_CERT / MQTT_TLS_KEY 后，QUIC 会在 9443/UDP 自动启用。前提是客户端支持 QUIC（paho / esp-mqtt / mosquitto 均不支持）。启用后需在防火墙开放 UDP 端口。

Q：消息追踪会影响性能吗？

未开启追踪时，每次消息额外开销约 1ns（无锁原子判断），完全可忽略。开启追踪后按匹配的消息数计，追踪 # + 高频场景可能下降 30-50% 吞吐。建议追踪时缩小 topic 范围并控制时长。

Q：统计弹窗会影响性能吗？

不打开统计弹窗时，publish 路径只做一次原子读（~1ns），不产生任何写操作。打开弹窗时，每条 PUBLISH 消息增加一次原子计数（~3ns），加上每 2 秒一次的统计快照读取（~30μs），相对 MQTT 消息处理（每条 1-10μs）完全可忽略。关闭弹窗 30 秒后自动恢复。

Q：为什么统计弹窗里消息吞吐量是 0？

两个原因：

• 没有活跃消息：没有客户端在发消息，速率自然是 0
• 用的是超级用户发消息：超级用户绕过 ACL hook，不会被计数

建议用普通用户发消息测试，且保持弹窗打开状态下发送。

Q：审计日志记录的是什么 IP？

访问 Web 面板的人的 IP（操作者），不是 MQTT 设备的 IP。想看 MQTT 设备 IP 请查询在线客户端列表的 remote_addr 字段。

Q：审计日志能存多少条？

SQLite 后端每 100 次插入清理一次，保留最近 100 万条；Redis 后端保留最近 10000 条（滑动窗口，超出自动淘汰最旧）。

Q：MySQL / PostgreSQL 落库和 GreptimeDB 落库能同时启用吗？

可以。三者独立运行，各自订阅主题、各自批量写入。同一个消息可以被写入多个目标。三个写入器使用不同的共享订阅组名（greptime_writer / mysql_writer / psql_writer），互不干扰。

Q：为什么落库写入器需要 mqhub-shared-subscription.toml？

落库写入器使用共享订阅（$share/...）。如果 mqhub-shared-subscription.toml 不存在，共享订阅不可用，写入器会启动失败并提示。这是为了避免静默降级成"每个节点都写入"导致数据重复。

Q：如何从 MQTT 消息里提取字段写入数据库？

使用 MySQL / PostgreSQL 的内容转换功能。开启后， payload 会被解析成结构化字段：

• JSON 提取：MYSQL_TRANSFORM_RULE=json_extract，从 JSON payload 里取字段
• 正则提取：MYSQL_TRANSFORM_RULE=regex，从文本里用正则提取

默认关闭（*_TRANSFORM_ENABLED=0），按需开启。 GreptimeDB 不支持转换——它是列式存储，需要结构化字段请用 MySQL / PostgreSQL。

Q：如何信任反向代理的 X-Forwarded-For？

设置环境变量 TRUST_PROXY=1。默认不信任，直接用 socket IP。开启后从 X-Forwarded-For 或 X-Real-IP 提取真实客户端 IP。

Q：如何开启匿名接入？

通过环境变量 MQTT_ALLOW_ANONYMOUS=1 开启，详见上文「匿名接入」章节。默认关闭。

Q：集群模式 3 节点还是 5 节点？

3 节点是最小可用集群，能容忍 1 节点故障；5 节点能容忍 2 节点故障，但写入延迟更高。3 节点是性能与成本的最佳平衡，只有容错要求极高的核心系统才考虑 5 节点。

Q：Raft 和 Broadcast 怎么选？

Raft：强一致性，防止 Client ID 重复登录，适合生产环境。Broadcast：无 Leader，最终一致性，吞吐量更高，适合海量传感器上报、能容忍短暂状态不一致的场景。

Q：4GB 机器能撑多少连接？

取决于消息频率、QoS 等级、是否启用 TLS 等。纯 TCP、无 TLS、低频率上报场景下，优化后的 MQHub 大约可以稳定承载 6 万 - 10 万 并发连接。建议在目标硬件上使用 rmqtt-bench 压测，关注 OOM 拐点。

Q：升级 MQHub 会不会丢失配置？

不会。所有配置都在环境变量中，所有业务数据都在 SQLite 或 Redis 里，替换二进制文件后直接启动即可。