上个月在做一个储能设备的接入项目，几百台设备分散在全国，每台两分钟上报一次电压、电流和温度。一开始用 HTTP 轮询，设备一多，服务端压力和数据延迟都压不住，后来整体切到 MQTT，问题才算真正解决。这篇文章把这次接入的完整思路写下来，从协议、broker 到代码，一条线讲完。
一、为什么物联网项目都绕不开 MQTT
MQTT 是基于 TCP 的发布订阅协议，专门为低带宽、高延迟、不稳定网络设计。设备不用管消息发给谁，只管往主题上发，broker 负责中转，谁订阅了谁收。HTTP 那种一问一答的模式，在设备端成了大问题，每台设备要频繁轮询或长连接，服务端扛不住，设备也费电费流量。
MQTT 真正厉害的地方在消息质量分层。QoS 0 最多一次，适合丢了也无所谓的告警日志；QoS 1 至少一次，消息必达但可能重复，需要消费端做幂等；QoS 2 恰好一次，代价最大，只在资金、订单这类场景用。大多数设备数据用 QoS 1 就够了，配合消费端去重，兼顾可靠和性能。
二、设备怎么连上来：broker 选型与部署
broker 是整个链路的中心，选型基本决定后面的运维体量。小项目用 Mosquitto 就够，单机、配置简单；设备量上千、要集群和高可用的，直接上 EMQX，开源版自带管理控制台、规则引擎和监控面板，还能对接 Webhook 把数据转发到业务服务。
部署上两个端口记清楚，1883 是明文端口，8883 是 TLS 加密端口。设备数据过公网必须走 TLS，不然设备位置、电量、控制指令全裸奔。证书用 EMQX 自带的配置项挂上去就行，设备端多配一个 CA 文件，代码层面几乎无感。
三、认证和权限，别让任何设备冒充你
设备接入最怕的一件事，是别人伪造一台设备连进来发假数据。MQTT 默认没有任何鉴权，谁拿到 broker 地址都能连，所以这一步必须自己做。
认证分三层。第一层是 clientId，每个设备全局唯一，重名连接会把老连接踢下线，天然防重复。第二层是用户名密码，EMQX 里可以按 clientId 绑定账号，每个设备一个独立的密码，泄露一台不影响其他设备。第三层是 ACL 权限，严格限制每台设备只能发布自己命名空间下的主题，比如 device_001 只能往 devices/device_001/# 发布，别的主题一概拒绝。
这套东西在 EMQX 控制台就能配，也可以走数据库动态加载，设备上线自动下发账号和权限。别图省事所有设备共用一个账号，出一次安全事故就够你后悔的。
四、主题、QoS、遗嘱，消息设计里的几个关键决定
主题结构直接影响后续扩展。常见做法是按 产品 / 设备 / 数据类型 三层组织，比如 devices/device_001/telemetry 放遥测数据，devices/device_001/command 放下行指令，devices/device_001/status 放上下线状态。订阅端用通配符，devices/+/telemetry 订阅所有设备，devices/+/status 只关注状态，一张主题表定下来，后面加设备类型不用改代码。
遗嘱消息和保留消息这两个特性，设备接入基本必用。遗嘱是设备异常断线时 broker 替它发的最后一条消息，配合 status 主题，服务端能立刻感知设备掉线。保留消息是 broker 存住最后一条，新订阅者一上来就能拿到当前状态，不用等设备再发一次。设备端配置就几行，价值非常大。
五、代码落地，设备端和服务端都能跑
设备端用 Node.js 的 mqtt 包，发布遥测数据，同时配好遗嘱和自动重连。
const mqtt = require('mqtt');
const client = mqtt.connect('mqtts://broker.example.com:8883', {
clientId: 'device_001',
username: 'device_001',
password: '每台设备独立的密码',
clean: false,
will: {
topic: 'devices/device_001/status',
payload: 'offline',
qos: 1,
retain: true
}
});
client.on('connect', () => {
client.publish('devices/device_001/status', 'online', { qos: 1, retain: true });
setInterval(() => {
client.publish('devices/device_001/telemetry', JSON.stringify({
voltage: 512.6,
current: 23.4,
temperature: 36.8,
ts: Date.now()
}), { qos: 1 });
}, 5000);
});服务端订阅所有设备的遥测，收到后解析入库。注意 QoS 1 会重复投递，入库前用 ts 和设备 id 做幂等，否则数据表里全是重复记录。
const mqtt = require('mqtt');
const client = mqtt.connect('mqtt://broker.example.com:1883', {
clientId: 'ingest_server',
username: 'service_account',
password: '服务端专用密码'
});
client.subscribe('devices/+/telemetry', { qos: 1 });
client.on('message', (topic, payload) => {
const deviceId = topic.split('/')[1];
const data = JSON.parse(payload.toString());
// 按 deviceId + ts 去重后写入时序库或业务库
});六、上线前再检查这几件事
第一件，心跳和重连参数。keepalive 设短一点，设备异常断线才能被快速发现，重连用指数退避，别所有设备同时挤上来。
第二件，broker 的共享订阅。多台服务端实例消费同一批设备消息时，用 $share/group/topic 分组，broker 自动把消息分给组内不同实例，比每台都订阅一份省事得多。
第三件，下行指令也要想清楚，控制类指令建议 QoS 1 加设备端应答，设备收到指令回一条确认，闭环才算成立。
MQTT 的坑大多不在协议本身，而在认证、主题和重连这些工程细节上。把这三件事做扎实，一万台设备连上来，服务端照样稳。
在线
电话
微信
需求
TOP
