MQTT实战 - QoS 0-1-2 到底怎么选？一张图看懂（四）-尧图手机网站定制

我是做物联网后端的老兵，8 年 .NET 开发，现在带一支研发团队做 .NET + AIoT 架构。这个系列讲我真实项目里踩过的坑——不是抄来的入门教程。

QoS 这三个字母，可能是 MQTT 里被误解最深的东西。新手要么一句「当然全选 QoS 2，最可靠」，要么一句「QoS 越高越费流量，能低就低」。这两种说法都不算错，但都没抓住本质——QoS 不是「可靠性档位」，而是你为一类业务支付的成本。

我将按「事故 → 原理 → 陷阱 → 代价 → 决策 → 场景 → 踩坑 → 落地 → 误区 → 速查」顺序详细讲解QoS到底应该如何选。

01、事故：两个极端，我都踩过

我约到过一个现场事故：平台下发了一条温控指令，设备没执行。排查了一下午，发现是QoS选错了——那条指令用的是QoS 0。QoS 0发出去就不管了，Broker不保证送达，设备离线或者网络抖动，消息直接丢。

后来我们在另一个项目里走了另一个极端——把所有消息统一设成QoS 2，觉得“最高级别肯定最保险”。结果设备量上来之后，Broker的CPU直接飙到90%，消息延迟从毫秒级变成了秒级。原因是QoS 2的四次握手开销太大，每一轮消息都要走PUBLISH → PUBREC → PUBREL → PUBCOMP，设备一多，Broker就被四次握手拖垮了。

还有一个智慧农业项目，2000个土壤传感器，每分钟上报一次温湿度，选的是QoS 1。我打开EMQX监控面板一看——消息重复率3.7%，部分设备甚至达到了8%。问开发为什么重复这么多，答：“我选的QoS 1啊，至少一次，保证不丢。”再问：“消费端做幂等了吗？”沉默。

选低了丢消息，选高了拖垮Broker，选中间不做幂等就重复消费。这就是QoS选型的核心矛盾。

我用三个月，把公司所有项目的QoS策略从“凭感觉”拉到了“按场景”。下面这套决策框架，就是那次整理的完整成果。

02、原理：QoS 0 / QoS1 / QoS2 核心机制拆解

MQTT 的 QoS 只解决一个问题：一条消息从发布者到订阅者，中间要确认几次。确认越多，越不容易丢，但开销越大、延迟越高。

下图为每个QoS等级的具体流程：

QoS 0｜最多一次（At most once）

报文流程：仅 PUBLISH 单向报文，无应答确认，发完即忘
投递保证：消息可能丢失，绝对不会重复
报文交互：1 次报文；开销最低、延迟最小
行为：发送方不缓存消息，无重传逻辑

✅适合：高频周期性采样，少量样本丢失业务可接受

❌不适合：告警、控制指令、不可丢失事件

QoS 1｜至少一次（At least once）

报文流程：PUBLISH → 等待PUBACK；超时未收到应答自动重传
投递保证：消息一定可达，但是会出现重复消息
报文交互：2 次报文；开销中等
行为：发送方缓存消息直到确认；网络抖动丢失 ACK 就会触发重传，这就是消息重复的根源

✅适合：告警、故障事件、平台下发控制指令；前提：业务实现幂等去重

❌不适合：计费计量等绝对不能重复执行的业务，直接裸用 QoS1 会引发事故

QoS 2｜恰好一次（Exactly once）

报文流程：四次握手 PUBLISH → PUBREC → PUBREL → PUBCOMP
投递保证：协议链路层面做到不丢失、不重复
报文交互：4 次报文；带宽、内存、延迟开销显著升高
行为：收发两端均维护 PacketID 会话状态；弱网环境握手极易超时，造成消息堆积、设备内存上涨

✅适合：计费结算、计量扣费，链路既不能丢也不能重复的少数业务场景

❌高危提醒：NB****IoT/2G/4G 弱网设备谨慎使用 QoS2，四次握手极易超时，会话积压引发设备掉线。绝大多数物联网业务优先采用QoS1 + 业务幂等替代 QoS2。

下面我们用一张表总结：

记住三句话就够：

QoS 0 = 发了就不管：丢了认栽，胜在快和轻。
QoS 1 = 保证送到，但可能重发：Broker 没回 PUBACK，重连后会把这条再发一遍——所以你有可能收到两条一模一样的。
QoS 2 = 保证只送到一次：四次握手去重，最稳，也最重。

03、陷阱：QoS 是"两端协商"的，不是"一端说了算"

记住一个规则：最终生效的 QoS = min(发布端 QoS, 订阅端授予的 QoS)。而且 QoS 是「分段协商」的——发布者 ↔ Broker 一段、Broker ↔ 订阅者一段，并不是发布者与订阅者端到端协商。你在发布端把 QoS 调到最高，订阅端没跟上，一样白搭。

真实惨案：一个团队排查大半天「为什么消息会重复」，最后发现是发布端 QoS 1、订阅端也 QoS 1——QoS 1 本来就允许重复，他们却一直以为 QoS 1 是「精确一次」。更危险的是把 QoS 2 当保险的人：发布端发了 QoS 2，订阅端只声明了 QoS 1，链路自动降级成「至少一次」，重复指令照样执行两次。所以端到端可靠性必须发布、订阅、Broker 三者一起对齐，缺一不可。

记住：QoS 是两个端点协商出来的，两端都要对齐语义，不能只管一头。

04、代价：选 QoS 要算的四笔账

选 QoS 不能只看「可靠性」，要把代价一起算清楚，否则很容易像开头那样顾此失彼。

关键认知：QoS 1 的「至少一次」意味着消息可能到达多次。如果消费端没有幂等保护，QoS 1 比 QoS 0 更危险——因为 QoS 0 丢了你能发现，QoS 1 重复了你未必能发现。此外，QoS 2 相对 QoS 0 的额外延迟约 33%、功耗高出约 92%，这对电池设备和弱网链路都是沉重负担。

05、决策：三问定 QoS（决策框架）

规则：从第一个问题开始，顺着箭头走，命中任何一个结论就直接停止判断；注意 QoS 是按「每条消息」独立指定的，不是整条连接一次性设定。

一句话总结：能容忍丢失用 QoS 0，不能容忍丢失但能处理重复用 QoS 1，不能容忍丢失也不能处理重复才考虑 QoS 2——而且 QoS 2 在弱网下是灾难。

06、场景：场景速查表

我按业务对「丢失 / 重复」的容忍度选型，以及配套幂等、版本号、Retain 等动作，整理了一份 MQTT 业务场景 QoS 选型建议 —— 可靠性按需升级，不必一律 QoS 2。

极简口诀：高频采样丢点无所谓 → QoS 0；消息不能丢允许重复 → QoS 1 务必做幂等；既不能丢又绝对不能重复且网络好 → QoS 2；NB-IoT/弱网设备尽量避开 QoS 2。

07、踩坑：5 个 QoS 惨案

惨案 1：周期遥测/温度上报 → QoS 0

踩过的坑：

早期温度上报用 QoS 1，每条都要等 Broker 的 PUBACK，射频模块反复唤醒，设备电池续航比预期短了近三分之一。另有项目 2000 台水表每 30 秒上报一次，全量使用 QoS 2，四次握手让 Broker 堆积上百万条“待确认”状态，CPU 飙到 95%。还有 2000 台设备每秒 QoS 1 上报，Broker 每秒处理 2000 个 PUBLISH + 2000 个 PUBACK，连接池被打满，新设备连不上。

规范：

传感器周期上报、温度等高频遥测默认用 QoS 0。丢了下一秒补上，不影响趋势分析。若业务要求读数可靠补传，参考惨案 2使用 QoS 1 + seq 幂等，不要用 QoS 2。

正确示例（MQTTnet）：

awaitclient.PublishAsync(newMqttApplicationMessageBuilder().WithTopic($"v1/{tenant}/sensor/{deviceId}/up/telemetry").WithPayload(payload).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtMostOnce)// QoS 0.Build());

原因：

QoS 0 功耗比 QoS 1 低约 27%，比 QoS 2 低约 92%；对电池供电设备，这直接决定续航是一个月还是一年。全量 QoS 2 会省掉 75% 的握手开销的反面代价——四次握手和待确认状态会压垮 Broker。

惨案 2：水表读数/NB-IoT 弱网需补传 → QoS 1 + seq 幂等

踩过的坑：

水表项目选 QoS 2 保证读数不重复。NB-IoT 延迟 500ms～2s，QoS 2 四步握手在弱网下超时重传，设备端单片机内存只有 64KB，去重缓存被撑爆，直接硬件重启。

规范：

弱网、低内存、需要可靠补传的读数上报，改用 QoS 1 + 读数序号 seq 做幂等；后端按 deviceId + seq 去重。不要用 QoS 2。

正确示例（MQTTnet）：

awaitclient.PublishAsync(newMqttApplicationMessageBuilder().WithTopic($"v1/{tenant}/meter/{deviceId}/up/reading").WithPayload(JsonSerializer.Serialize(new{seq,// 每帧读数序号reading,ts=DateTimeOffset.UtcNow})).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)// QoS 1.Build());// 后端幂等：按 deviceId + seq 去重，已处理则丢弃

原因：

QoS 1 保证“至少一次”，重复由业务层 seq 幂等兜底；效果接近 QoS 2，但开销低得多，也不会在 64KB 内存设备上撑爆去重缓存。

惨案 3：平台下发控制指令 → QoS 1 + cmdId 幂等

踩过的坑：

智能照明项目平台下发“开灯”指令用 QoS 1，网络抖动导致 PUBACK 丢失，设备重连后 Broker 重发，灯被开了两次，设备日志出现两次执行记录，审计不通过。也有控制指令用 QoS 0 导致指令丢失；重复执行“开启阀门”还可能造成机械故障。

规范：

控制指令用 QoS 1，确保至少送达一次；指令必须携带唯一 cmdId（UUID），设备侧维护已处理指令 ID 的短时缓存，收到指令先查 cmdId 是否已处理，处理过则跳过。

正确示例（指令带唯一 ID + 设备侧幂等）：

// 平台下发指令，每条指令带唯一IDvarcommandId=Guid.NewGuid().ToString("N");awaitclient.PublishAsync(newMqttApplicationMessageBuilder().WithTopic($"v1/{tenant}/hvac/{deviceId}/down/cmd").WithPayload(JsonSerializer.Serialize(new{commandId,action="open_valve",timestamp=DateTimeOffset.UtcNow})).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)// QoS 1.Build());// 设备侧：维护已处理指令ID的短时缓存（Redis或本地内存）// 收到指令先查 commandId 是否已处理，处理过则跳过

原因：

QoS 1 保证“至少一次”，但不保证“不重复”。MQTT 规范里 QoS 1 的重传发生在客户端重连时，连接断开后未收到 PUBACK 的消息会重新发送，重复不可避免，必须靠业务层幂等兜底。QoS 1 + 幂等效果等同 QoS 2，但开销低得多。

惨案 4：故障告警 → QoS 1 + eventId 幂等

踩过的坑：

化工厂气体检测项目，浓度告警用 QoS 0。一次网络闪断，告警丢了，现场浓度已经超标但平台没收到通知，差点出人命。另一个项目把“设备高温告警”设成 QoS 0，设备温度到 80 度告警消息丢失，等运维巡检才发现设备已经烧了。

规范：

告警消息用 QoS 1，且告警消息带 eventId，消费端做幂等。只有“重复执行会造成真实伤害”的场景（计费、金融交易、紧急停机）才考虑 QoS 2。

正确示例（QoS 1 告警）：

vareventId=Guid.NewGuid().ToString("N");awaitclient.PublishAsync(newMqttApplicationMessageBuilder().WithTopic($"v1/{tenant}/device/{deviceId}/up/event").WithPayload(JsonSerializer.Serialize(new{eventId,// 告警唯一ID，消费端幂等eventType="high_temperature",severity="critical",value=82.5,ts=DateTimeOffset.UtcNow})).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.AtLeastOnce)// QoS 1.Build());

原因：

QoS 1 送达率实测可达 99.8%，对告警场景足够；重复推送告警不会造成伤害——收到两条“高温告警”和收到一条，处理逻辑一样。QoS 2 送达率接近 100%，但延迟增加约 33%，对告警场景，多出来的延迟比那 0.2% 的丢包更危险。

惨案 5：计费/扣款/金融交易/紧急停机 → QoS 2 + 业务幂等

踩过的坑：

智能充电桩计费一开始用 QoS 1，偶尔出现设备重复上报“充电结束”事件，导致用户被重复扣费，后来改成 QoS 2。另有金融物联网项目选 QoS 2 后觉得“恰好一次，不用管幂等”，结果 Broker 集群切换时部分消息在切换窗口内被投递了两次。

规范：

计费、金融交易、紧急停机这类“丢不起也重不起”的场景用 QoS 2；但关键业务无论选什么 QoS，消费端都必须做业务幂等。QoS 是传输层保证，不是业务层保证。

正确示例：

awaitclient.PublishAsync(newMqttApplicationMessageBuilder().WithTopic($"v1/{tenant}/charger/{chargerId}/up/billing").WithPayload(JsonSerializer.Serialize(new{sessionId,amount=12.50,timestamp=DateTimeOffset.UtcNow})).WithQualityOfServiceLevel(MqttQualityOfServiceLevel.ExactlyOnce)// QoS 2.Build());

原因：

QoS 2 通过两阶段确认（PUBLISH/PUBREC + PUBREL/PUBCOMP）确保消息既不丢失也不重复。但代价是开销显著增加——QoS 2 功耗比 QoS 0 高约 92%，延迟增加约 33%。所以 QoS 2 只能用于“丢不起也重不起”的极少数场景，绝不要全量使用；同时 Broker 之间的 QoS 2 保证不等于端到端恰好一次，消费端仍要按业务 ID 幂等。

08、落地：在 .NET（MQTTnet）里怎么落地

QoS 1 的幂等处理

既然 QoS 1 是大多数场景的最优解，幂等就是必修课。我们用的是「消息去重表 + 业务幂等」，三种常见方案按场景挑。

方案一：Redis 去重表（推荐）

每条消息携带 messageId，消费端用 SET NX 判断是否已处理；过期时间要大于 Broker 的重传周期，我们的经验值是 5 分钟。

publicclassMessageDeduplicator{privatereadonlyIDatabase_redis;privatereadonlyTimeSpan_expiry=TimeSpan.FromMinutes(5);publicMessageDeduplicator(IConnectionMultiplexerredis)=>_redis=redis.GetDatabase();// 返回 true 表示是新消息；false 表示重复消息publicTask<bool>TryMarkAsProcessedAsync(stringdeviceId,stringmsgId)=>_redis.StringSetAsync($"mqtt:dedup:{deviceId}:{msgId}","1",_expiry,When.NotExists);}// 使用：先查重，再处理varmsgId=ExtractMessageId(payload);if(!awaitdedup.TryMarkAsProcessedAsync(deviceId,msgId)){logger.LogDebug("Duplicate skipped: {D}/{M}",deviceId,msgId);return;}awaitHandleAsync(payload);

方案二：业务唯一键去重

利用业务本身的唯一性，比如 deviceId + 时间戳（精确到秒）+ 数据类型，用数据库唯一索引或 Redis 去重：

vardedupKey=$"{msg.DeviceId}:{msg.Timestamp:yyyyMMddHHmmss}:{msg.Type}";// 用 dedupKey 做数据库唯一索引或 Redis 去重

方案三：版本号 / 序列号

设备端维护自增 seq，后端记录每个设备的 lastSeq，小于 lastSeq 的消息直接丢弃：

if(msg.Seq<=lastSeq[msg.DeviceId])return;// 重复或乱序，直接丢弃lastSeq[msg.DeviceId]=msg.Seq;

注意两点：去重表只对「控制指令」和「状态变更」类消息启用，传感器上报数据不需要去重；如果业务本身天然幂等，比如「设置温度阈值为 25 度」，执行两次和执行一次结果一样，也可以不走去重表——但「累加计数」「扣款」这类非幂等操作，必须走去重。

语义化 QoS 策略（防写错）处理

光记住规则还不够，得让代码帮你守规矩。我在项目里把 QoS 按业务语义固化成策略，绝不随手写数字：

usingMQTTnet;usingMQTTnet.Protocol;/// <summary>/// 按业务类型的 QoS 策略：不同消息走不同档位/// </summary>publicstaticclassQosPolicy{// 高频遥测：允许至多一次，省带宽省电publicstaticMqttQualityOfServiceLevelTelemetry=>MqttQualityOfServiceLevel.AtMostOnce;// QoS 0// 控制指令 / 告警：至少一次，绝不能丢publicstaticMqttQualityOfServiceLevelCommand=>MqttQualityOfServiceLevel.AtLeastOnce;// QoS 1// 计费 / 结算：恰好一次，不丢不重publicstaticMqttQualityOfServiceLevelBilling=>MqttQualityOfServiceLevel.ExactlyOnce;// QoS 2}// 发布遥测数据（QoS 0）awaitmqttClient.PublishAsync(newMqttApplicationMessageBuilder().WithTopic("v1/t01/energy/meter/vdev001/up/telemetry").WithPayload(payload).WithQualityOfServiceLevel(QosPolicy.Telemetry).WithRetainFlag(true)// 状态类消息才 Retain.Build());// 下发控制指令（QoS 1）awaitmqttClient.PublishAsync(newMqttApplicationMessageBuilder().WithTopic("v1/t01/energy/meter/vdev001/down/cmd").WithPayload(cmdPayload).WithQualityOfServiceLevel(QosPolicy.Command).Build());