智能售货新潮流SpringBootMQTT打造取货即走黑科技最近在做一个智能售货项目核心就一句话用户扫码开门拿走商品关门自动扣款全程不用在手机上一顿操作。之所以叫它“取货即走”黑科技是因为它把传统售货机的“选货—扫码—支付—出货”四个动作压缩成了一个动作——关门。这套系统的技术底座很简单后端用SpringBoot承担订单、库存、支付对账这些正经业务设备端到云端这一截则全部交给MQTT协议。写这篇文章的时候我已经在真实的柜体上跑过了两轮灰度从原型到量产踩了不少坑这篇就把整个项目的思路、协议设计、核心代码和排障过程都拿出来聊聊。适合正在做无人零售、共享设备、IoT后端或者想搞清楚MQTT项目中那些弯弯绕绕的朋友参考。1. 为什么是SpringBootMQTT从技术选型说起1.1 设备通信为什么不走HTTP先回答一个很基础的问题售货机上报状态、接收开锁指令为什么不用HTTP轮询非要引入MQTTHTTP轮询的方案我也纠结过一阵子。设备每3秒上报一次心跳服务端有指令下发时只能等设备下次上报时带回去或者靠长连接推送。遇到售货机这种强实时场景——用户扫码后柜门必须在1秒内弹开——HTTP轮询的延迟就成了硬伤。3秒一次的轮询周期运气不好用户要等两秒多才开门这在产品体验上是不可接受的。把轮询间隔压到500ms又会让服务端压力暴涨1000台设备每分钟就要处理12万次心跳成本全耗在无意义的网络往返上。MQTT解决这个问题的思路是“推”而不是“拉”。设备与Broker之间建立一条常驻的TCP长连接服务端下发指令时消息能够实时到达设备端中间没有轮询等待。加上MQTT的发布/订阅模型天然适合“一台设备对应多个业务方”的场景设备状态可以同时被订单服务、告警服务、数据大屏订阅不需要业务方各自去设备拉取。后面我跟硬件同事聊过设备端用ESP32或者工业级4G模组跑MQTT的开销也远低于HTTPJSON的解析成本整套方案在功耗和流量费上都有明显优势。1.2 SpringBoot在后端体系里的定位设备通信这半边交给MQTT但业务这半边我仍然全部用SpringBoot来实现。SpringBoot在这个项目里的角色不仅仅是“HTTP接口框架”它同时承担了三块职责对外提供小程序/App调用的REST接口扫码、查询订单、发起支付对内作为MQTT消息的消费者处理设备上报事件门状态变更、重力传感器变化、温度异常以及跑定时任务做订单超时和库存巡检。一台服务把设备接入、业务处理、运营后台全部包圆对于几十台到几百台设备的项目规模来说完全够用不需要一上来就拆微服务。为什么不用Netty自研长连接或者WebSocket方案原因在于MQTT协议本身已经把设备接入的很多脏活累活干完了断线重连、心跳保活、消息QoS、遗嘱清理这些都是协议层写好的直接用Broker我选的是EMQX就能拿到稳定可靠的连接管理能力。如果用Netty自研这些机制都要自己实现一遍开发周期翻倍不说稳定性还得靠时间验证。SpringBoot在这里的价值是让我能用最熟悉的Java生态把业务逻辑快速落地借助Spring的依赖注入和事件机制把设备消息、订单状态、支付回调优雅地串起来。1.3 技术栈全景与架构分层我把这套系统的技术栈列个清单方便你对照自己的项目评估层级组件说明设备端ESP32 / 4G DTU MQTT客户端采集门锁状态、重力传感器、温控数据接入层EMQX Broker设备连接管理、主题转发、遗嘱消息处理业务层SpringBoot Web应用订单服务、设备管理、库存扣减、支付回调消息层Spring Integration MQTT订阅设备消息并转成Spring事件数据层MySQL Redis订单数据落库库存与令牌用Redis用户端微信小程序扫码、查看商品、支付结果展示全局部署上EMQX和SpringBoot同机部署之间走内网。设备通过4G网络连接EMQX的1883端口MQTT over TCP如果有条件可以加上TLS走8883但考虑到设备端算力有限我在量产机型上用的是TCP明文加上应用层加密。这里说明一下售货柜场景里消息内容本身不涉及用户隐私字段主要是设备ID、订单号、状态码这些因此应用层做了个简单的字段级签名校验避免伪造消息注入。2. MQTT主题与消息设计不丢单的核心功夫2.1 Topic规划三个层级定全局MQTT项目里第一个要紧的设计就是Topic。我在第一个版本里Topic起得乱七八糟设备上报、下发的路径混在一起后来排查线上问题找不到对应消息才狠下心来重新设计了整套主题体系。现在这个项目里所有Topic都遵循三条规范一是命名按“域/设备类别/设备标识/动作”四段式组织二是上行设备到服务端和下行服务端到设备分开三是不在Topic里放业务订单号这些随机字段订单相关信息一律放在消息体里。实际用到的核心主题如下vending/cmd/{deviceId}/open服务端下发开锁指令消息体包含订单号、超时时间vending/cmd/{deviceId}/close强制锁门管理后台远程操作vending/event/{deviceId}/state设备上报门状态、传感器数据这个是最高频的消息vending/event/{deviceId}/alarm设备上报异常告警比如门未关、重力异常vending/event/{deviceId}/heartbeat设备心跳包含电量、信号强度、固件版本为什么要按设备ID分Topic而不是所有设备共用一个Topic一开始我觉得共用一个Topic多省事服务端订阅一次就能收到所有设备的消息。但这样做的两个问题很快暴露出来一是没法利用MQTT本身的Topic权限做设备级隔离一台设备理论上能收到其他设备的信息二是调试的时候极其痛苦抓日志根本分不清消息来自哪台设备。按设备分Topic之后每个设备只能订阅自己的指令主题服务端也能按设备维度做消息追踪遇到问题一次就能定位到具体设备。2.2 QoS选型QoS 1是最优解MQTT有三档QoS我直接说结论这个项目里所有业务消息都用了QoS 1只有心跳用了QoS 0。为什么不用QoS 2QoS 2虽然能保证消息不重不漏但它需要四段握手每一条消息的确认开销很大用在售货柜这种高频状态上报场景下吞吐量上不去。实际业务里“重复”并不可怕我把消息处理做成了幂等——真正可怕的是“丢失”尤其是开锁指令丢了用户会被锁在柜子前面。QoS 1保证消息至少到达一次但会把重复消息的问题留给你。我在设备端和服务端都做了去重处理每一条业务消息都带一个自增的消息序号或业务单号接收方在Redis里维护一个最近N条消息ID的滑动去重集合重复的消息直接丢弃。Redis的SETNX配合过期时间特别好用比在内存里维护HashSet要可靠得多服务重启后去重状态也不会丢。关于Retain消息我只在设备启动时的初始状态同步里用了它。设备连接EMQX后向其状态Topic发布一条带Retain标志的当前状态快照这样即使服务端在设备离线期间错过了某些状态变化新上线的服务端订阅也能立刻拿到设备的最新状态不会出现“服务起来了但不知道设备什么状态”的尴尬。业务命令类消息一律不设Retain避免设备重连后收到一条几小时前的开锁指令那会引发严重的安全问题。2.3 遗嘱消息给设备“死亡”一个准确的信号在线状态判断是所有IoT项目都会遇到的问题。HTTP时代我们靠心跳超时判离线比如连续10个周期没收到心跳就认为设备挂了。但心跳超时有个问题服务端需要等待一段时间才能确认设备离线而且如果设备网络波动导致心跳延迟很容易误判。MQTT的遗嘱LWT机制从协议层面解决了这个问题——设备建立连接时带上遗嘱消息如果设备异常断开比如断电、断网Broker会立刻替设备把这条遗嘱消息发出去。我在设备上线时设置遗嘱内容为{online:false,deviceId:xxx,ts:...}发布到设备状态主题正常下线时设备会主动发一条online:false的消息并断开连接Broker不会再发遗嘱。这样服务端能在秒级内感知设备离线及时把设备标记为不可售。这里有个容易踩的坑设备SDK在设置遗嘱时一定要把遗嘱的Topic和正常上报状态的Topic保持一致否则服务端要订阅两个主题才能感知在线状态极容易漏消息。真实项目里出现过设备断网后App还在正常售货因为服务端没收到离线信号直到用户下单后开不了门才暴露问题。2.4 消息格式JSON不是性能瓶颈别过度设计消息体我用的是JSON而不是二进制或Protobuf。我知道很多IoT团队会强调二进制协议的各路优势但在这个项目里我坚定选择了JSON原因有三个第一售货柜消息体只有几十到几百字节JSON的解析开销完全可接受第二调试的时候mosquitto_sub订阅主题打印出来的消息直接可读不用写解码器第三硬件团队用的ESP32上有成熟的ArduinoJson库解析成本也在可控范围内。对于需要更低功耗的LoRa或NB-IoT场景我会推荐Protobuf但在4G网络环境下的售货柜JSON是投入产出比最高的选择。消息体我始终保持统一的信封格式{ msgId: uuid生成或者设备端递增id, deviceId: dv202406130001, type: DOOR_STATE, data: { door: OPEN, weight: 1.2, ts: 1718260000000 } }msgId是全链路幂等的关键字段服务端收到消息后先查这个Id是否处理过处理过直接返回。type字段用来标识消息类型避免一个Topic承载多种类型时的解析混乱。data里才是实际业务数据。这套信封格式从第一版沿用到现在后续扩展新消息类型时只增加type值服务端的事件分发逻辑完全不用改。3. SpringBoot服务端核心实现消息接入与业务落地3.1 依赖引入与核心配置SpringBoot整合MQTT我用了Spring Integration的spring-integration-mqtt它在MqttPahoClientFactory的基础上提供了消息驱动通道MqttInbound订阅到的消息会自动转成Spring的Message对象进入服务端管道。依赖引入如下dependency groupIdorg.springframework.integration/groupId artifactIdspring-integration-mqtt/artifactId /dependency dependency groupIdorg.eclipse.paho/groupId artifactIdorg.eclipse.paho.client.mqttv3/artifactId version1.2.5/version /dependencyapplication.yml里的关键配置项mqtt: broker: tcp://127.0.0.1:1883 client-id: vending-server-${random.value} username: vending password: xxx topic-pattern: vending/event/ server-qos: 1 keepalive: 60 connection-timeout: 10 max-inflight: 1000几个配置项的说明client-id必须唯一如果两个客户端用相同ID连接同一个BrokerEMQX会强制把旧连接踢掉这在生产环境会造成服务端反复断连keepalive我设60秒比设备端的120秒短这样Broker能及时感知服务端异常max-inflight控制未确认消息数量上限对QoS 1来说就是允许有多少条消息同时处于“已发送未确认”状态设备上报密集时这个值调大能减少阻塞。3.2 消息接入管道从MQTT到Spring事件Spring Integration把MQTT接入的方式是定义FlowMQTT入站适配器收到消息后交给处理器处理器解析信封并发布Spring事件。我在代码里把这件事做得尽量简洁清爽Bean public IntegrationFlow mqttInboundFlow( MqttPahoClientFactory factory, MessageHandler mqttMessageHandler) { return IntegrationFlows .from(new MqttPahoMessageDrivenChannelAdapter( vending-server, factory, vending/event/)) .handle(mqttMessageHandler) .get(); }关键的还是消息处理器的实现。这里我做了三件事解析Topic提取deviceId解析消息体做幂等校验然后按type字段分发到不同业务方法Override public void handleMessage(Message? message) { String topic (String) message.getHeaders().get( IntegrationMessageHeaderAccessor.HEADER_TOPIC); String deviceId parseDeviceId(topic); String payload (String) message.getPayload(); MqttEnvelope envelope JsonUtils.parse(payload); if (envelope null) return; if (!this.idempotentService.tryProcess(envelope.getMsgId())) return; switch (envelope.getType()) { case DOOR_STATE: this.deviceStateService.handleDoorState(deviceId, envelope.getData()); break; case ALARM: this.alarmService.handleAlarm(deviceId, envelope.getData()); break; case HEARTBEAT: this.deviceService.refreshHeartbeat(deviceId, envelope.getData()); break; default: log.warn(unknown message type: {}, envelope.getType()); } }IdempotentService是纯Redis实现setIfAbsent幂等标识过期时间设30秒——覆盖一条消息从入站到业务处理完成的耗时比时间短了可能会漏防重长了Redis内存备受压。批量开锁高峰期这个条件判断能兜住很大一部分重复消息吞吐量上有明显感知。3.3 订单中心的闭环设计“取货即走”的核心订单流程是用户扫码开门创建预订单 → 取货关门上报重量变化生成扣款订单 → 支付回调微信支付免密扣款 → 通知用户。我重点说说扣款金额怎么算。售货柜用的是重力传感器闭门后会上报一个重量值服务端拿这个重量减去初始货盘重量得到“减少的重量”然后匹配商品克重计算出实际拿走的商品数量和应付金额。这里有个细节同一种商品SKU的重量是固定的但传感器存在温漂和机械误差我做了2%的标准差容错。如果计算出来的数量有多解比如500g的结果可能是1瓶500ml的水也可能是2袋250ml的饮料系统会走人工介入流程——先把订单挂在待确认状态推送运营后台由人工确认扣款金额。这个兜底逻辑虽然看起来不“黑科技”但真实项目里没有它监控会接到大量投诉电话。订单状态机我用了最经典的待支付→支付中→已支付→已退款加上各种异常态。Redis贯穿整条链路预创建订单用Redis存了15分钟有效期的柜门令牌用户关门扣款时用Redis分布式锁防并发最后MySQL负责订单数据落库。简单说Redis管热数据MySQL管冷数据各自干擅长的事。4. 端到端业务流程从扫码开门到自动结算的全链路4.1 一次完整的“取货即走”时序下面是我整理的完整时序逻辑你对照自己项目的流程看看有没有落下关键环节用户在小程序扫柜门二维码小程序调服务端POST /api/v1/orders/prepare传设备ID和用户身份。服务端校验设备在线状态与售货状态生成预订单和15分钟有效的“开门令牌”通过MQTT向设备主题vending/cmd/{deviceId}/open下发开锁指令。设备收到指令后打开电磁锁向状态主题上报DOOR_STATE为OPEN服务端收到后更新预订单状态为“柜门已打开”开始计时。用户拿走商品关上柜门。设备检测到门锁到位等待重力传感器读数稳定后上报本次重量变化。服务端收到闭门消息此时订单中心开始算账重量差→商品匹配→金额确定。调用微信免密支付接口完成扣款。扣款成功订单状态置为“已支付”通过小程序订阅消息推送扣款明细扣款失败订单转待补缴锁定该用户后续补缴完成前无法再次使用。如果用户15分钟内未关门比如开了门啥都没拿就走了预订单自动取消服务端下发锁门指令并记录一次“超时未关”事件。这套流程最关键的体验点在于第5步和第6步之间的时间窗口。用户吃完饭关门就想看扣款结果结果服务端要等传感器数据稳定又要跑商品匹配还要调支付整个流程超过3秒就会出现大量客服咨询。我实测下来稳定状态下闭门到支付完成平均耗时1.8秒主要开销在传感器稳定周期约1秒和支付回调约0.5秒剩下的都是纯计算时间。4.2 超时、崩溃与异常兜底系统崩溃这件事在最坏的情况下会发生用户关了门服务端掉电了传感器数据没来得及处理。订单就这么挂在“门已打开”状态用户也走了柜门还锁着商品少了钱没扣到。针对这种情况我设计了三个兜底机制。第一个是“状态补偿扫描”定时任务每5分钟扫描一次“门已打开”状态超过10分钟的预订单主动向设备查询当前闭门状态和重量读数如果设备已关门则补跑结算流程。第二个是“异常订单看板”所有超过3分钟未结算到已支付状态的订单实时推送到运营后台人工可以一键介入补单或发起退款流程。第三个是“断电恢复补报”设备端固件上有本地断电续传逻辑设备重启后会把未上报的门状态和传感器记录补发一遍兜住设备断电后丢消息的最极端场景。这些兜底机制单独拎出来都很简单但组合起来才能形成一个基本闭环的无人零售系统。我给项目做故障注入测试的时候发现总体丢单率从第一版的5%降到了稳定运行后的0.1%以下这0.1%基本都发生在设备侧传感器彻底损坏的物理故障场景这部分只能靠定期巡检去覆盖了。4.3 WebSocket推送让用户实时看到扣款结果支付结果要通过小程序通知用户我选择了WebSocket实时推送而不是传统的小程序订阅消息。原因很简单用户关门的瞬间人还在柜子面前小程序页面正停留在订单详情页WebSocket推过来的扣款提醒直接刷新当前页面体验上是“关了门就看到钱扣了多少”而不是等一个系统通知。我在SpringBoot里用WebSocketHandler维护了一个userId → Session的映射表把扣款结果通过JSON直接推给前端。设备上报→结算→WebSocket推送整条链路在500ms内完成用户体感几乎是实时的。WebSocket只在用户打开小程序页面时建立页面关闭连接就断开所以对服务端的长连接压力在可控范围内。测试下来单机几千并发在线没有明显性能问题。如果后续用户量上来了横向加机器部署WebSocket集群时要注意Session共享问题那又是个新故事了本文先不展开。5. 典型故障与排查技巧实录5.1 消息重复导致重复扣款上线第一周就翻过车用户拿了一瓶水扣了两次款。排查发现是网关设备在信号弱的时候把QT消息重发了一次设备端没有处理好去重逻辑服务端收到的msgId相同但我的幂等服务竟然没拦住。为什么会没拦住因为我把幂等过期时间设成了30秒而网关重发消息时已经过去了40秒。在弱网环境下MQTT QoS 1的重发可能延迟很久远大于30秒的窗口。这个问题给了我一个教训幂等窗口不能拍脑袋定要看业务链路的最大允许延迟。最后我把幂等过期时间调整为10分钟同时把订单支付环节也做了数据库层的唯一约束——一个订单只能有一条成功支付流水。两层防线再也没出现过重复扣款的投诉。5.2 断线重连风暴打爆Broker第二批机器上线的时候出现过一次EMQX CPU打满的情况。排查后发现是运营商在凌晨4点做网络割接所有4G设备同时断开重启后全部在极短时间内尝试重连EMQX承受不住几千个连接同时建立的握手压力直接假死殃及所有在线设备。解决办法有两层。第一层是在EMQX配置里打开了zone.max_connections限流和自动背压第二层更关键在设备端固件里做了“重连退避策略”——第一次断线立即重连重连失败后等待随机2到5秒之后每次等待时间按指数增长最多不超过5分钟并且每次重传前加上设备ID的哈希散列让所有设备错峰连接。这个策略上线后即便是大规模断网复网EMQX的CPU峰值也稳稳控制在30%以内。5.3 时序错乱先收到闭门后收到开门MQTT的QoS保证的是单条消息的送达不保证消息顺序。实际运行中我遇到过一种诡异情况设备上报DOOR_CLOSE的消息竟然比DOOR_OPEN先到达服务端。虽然发生概率极低但一旦发生订单中心就会拿错误的重量数据去扣款。处理这个问题的思路是“服务端不做顺序假设用状态机校验”。设备上报状态变化时服务端检查当前存储的设备状态如果从“已关门”直接跳变到“已关门”这个变化和已知状态矛盾判定消息乱序丢弃无效消息并标记一个设备侧异常等待下一次心跳状态对齐。同时设备端在固件里给状态消息加了递增序号服务端发现有消息序号断层时主动发起状态同步请求而不是盲目处理乱序消息。5.4 用MQTT Explorer做日常排障的效率飙升处理MQTT项目强烈建议装一个MQTT Explorer做日常调试和排障。它能可视化订阅多个主题实时看到每个主题的消息流还能用类似SQL的过滤器筛选特定设备ID的消息。每次运营反馈“某台设备异常无操作”我都能通过MQTT Explorer订阅该设备的所有主题查看最近的上报记录和指令下发记录在几分钟内判断是设备掉线、指令没下发还是服务端处理出错不需要翻一堆分散的日志。我日常排障的固定操作流程是这样的MQTT Explorer订阅vending/event/#和vending/cmd/#过滤目标设备ID查看设备有没有上报心跳没有则先确认设备和网络状态有上报但服务端没回应看服务端日志里有没有报错或幂等拦截定位到服务端处理的断点后再结合Redis里的订单状态判断具体卡在哪个环节。这套流程在四个故障案例里都帮我快速定位了问题从原来的半小时排查时间缩短到10分钟以内。5.5 性能与容量一台服务器能扛多少台设备最后聊一个运营经常问的问题这套系统一台服务器能扛多少台设备我以一台4核8G的云服务器、EMQX和SpringBoot同机部署为例给出一个实测参考数据EMQX侧稳定支撑3000台设备的连接和消息转发没有问题这时SpringBoot的处理速度会成为瓶颈瓶颈的主要来源是订单结算时对Redis和MySQL的访问开销。如果每台设备平均每30秒上报一条状态消息一秒约100条消息进入业务处理4核机器CPU跑在70%左右。要扩容时优先把EMQX单独拆出去部署再给MySQL加从库分担读流量这样把单机支撑量提升到5000台以上没有太大压力。关于SpringBoot的线程池这里有一个值得注意的经验MQTT入站的消息处理是阻塞的如果把消息接入直接跑在Tomcat的工作线程里一旦业务处理中有数据库慢查询整个消息通道都会被拖慢。标准的做法是给消息处理单独配一个线程池控制并发数并且使用独立的数据库连接池资源避免消息处理把请求线程池打满影响REST接口的响应。我用了ThreadPoolTaskExecutor核心线程数设8、最大16、队列容量2000实测效果良好。6. 给想做类似项目的朋友几句真心话这套系统从需求确认到第一期落地前后大约三个月。回头看技术选的并不复杂——SpringBoot和MQTT都是各自领域里非常成熟的东西真正花时间的是边界场景的打磨弱网重传、异常状态恢复、重复消息兜底、金额计算容错没有一个是可以跳过的“锦上添花”。奉劝想快速复刻一个“取货即走”项目的朋友把60%的时间留在业务闭环和异常处理上而不是一味追求炫酷的设备端特效和新框架的酷炫用法。几个关键建议再重复一遍Topic规划必须一开始做全局设计不要等设备量大了再回头改一切设备消息必须带上全局唯一ID这是后续所有幂等和排查工作的前提流程设计上永远要考虑服务端掉电、设备断网、用户乱操作这三类“故障场景”。把这三件事想透你的项目大概率能比市面上很多售货方案稳得多。
SpringBoot+MQTT智能售货柜：取货即走系统架构与排障实战
智能售货新潮流SpringBootMQTT打造取货即走黑科技最近在做一个智能售货项目核心就一句话用户扫码开门拿走商品关门自动扣款全程不用在手机上一顿操作。之所以叫它“取货即走”黑科技是因为它把传统售货机的“选货—扫码—支付—出货”四个动作压缩成了一个动作——关门。这套系统的技术底座很简单后端用SpringBoot承担订单、库存、支付对账这些正经业务设备端到云端这一截则全部交给MQTT协议。写这篇文章的时候我已经在真实的柜体上跑过了两轮灰度从原型到量产踩了不少坑这篇就把整个项目的思路、协议设计、核心代码和排障过程都拿出来聊聊。适合正在做无人零售、共享设备、IoT后端或者想搞清楚MQTT项目中那些弯弯绕绕的朋友参考。1. 为什么是SpringBootMQTT从技术选型说起1.1 设备通信为什么不走HTTP先回答一个很基础的问题售货机上报状态、接收开锁指令为什么不用HTTP轮询非要引入MQTTHTTP轮询的方案我也纠结过一阵子。设备每3秒上报一次心跳服务端有指令下发时只能等设备下次上报时带回去或者靠长连接推送。遇到售货机这种强实时场景——用户扫码后柜门必须在1秒内弹开——HTTP轮询的延迟就成了硬伤。3秒一次的轮询周期运气不好用户要等两秒多才开门这在产品体验上是不可接受的。把轮询间隔压到500ms又会让服务端压力暴涨1000台设备每分钟就要处理12万次心跳成本全耗在无意义的网络往返上。MQTT解决这个问题的思路是“推”而不是“拉”。设备与Broker之间建立一条常驻的TCP长连接服务端下发指令时消息能够实时到达设备端中间没有轮询等待。加上MQTT的发布/订阅模型天然适合“一台设备对应多个业务方”的场景设备状态可以同时被订单服务、告警服务、数据大屏订阅不需要业务方各自去设备拉取。后面我跟硬件同事聊过设备端用ESP32或者工业级4G模组跑MQTT的开销也远低于HTTPJSON的解析成本整套方案在功耗和流量费上都有明显优势。1.2 SpringBoot在后端体系里的定位设备通信这半边交给MQTT但业务这半边我仍然全部用SpringBoot来实现。SpringBoot在这个项目里的角色不仅仅是“HTTP接口框架”它同时承担了三块职责对外提供小程序/App调用的REST接口扫码、查询订单、发起支付对内作为MQTT消息的消费者处理设备上报事件门状态变更、重力传感器变化、温度异常以及跑定时任务做订单超时和库存巡检。一台服务把设备接入、业务处理、运营后台全部包圆对于几十台到几百台设备的项目规模来说完全够用不需要一上来就拆微服务。为什么不用Netty自研长连接或者WebSocket方案原因在于MQTT协议本身已经把设备接入的很多脏活累活干完了断线重连、心跳保活、消息QoS、遗嘱清理这些都是协议层写好的直接用Broker我选的是EMQX就能拿到稳定可靠的连接管理能力。如果用Netty自研这些机制都要自己实现一遍开发周期翻倍不说稳定性还得靠时间验证。SpringBoot在这里的价值是让我能用最熟悉的Java生态把业务逻辑快速落地借助Spring的依赖注入和事件机制把设备消息、订单状态、支付回调优雅地串起来。1.3 技术栈全景与架构分层我把这套系统的技术栈列个清单方便你对照自己的项目评估层级组件说明设备端ESP32 / 4G DTU MQTT客户端采集门锁状态、重力传感器、温控数据接入层EMQX Broker设备连接管理、主题转发、遗嘱消息处理业务层SpringBoot Web应用订单服务、设备管理、库存扣减、支付回调消息层Spring Integration MQTT订阅设备消息并转成Spring事件数据层MySQL Redis订单数据落库库存与令牌用Redis用户端微信小程序扫码、查看商品、支付结果展示全局部署上EMQX和SpringBoot同机部署之间走内网。设备通过4G网络连接EMQX的1883端口MQTT over TCP如果有条件可以加上TLS走8883但考虑到设备端算力有限我在量产机型上用的是TCP明文加上应用层加密。这里说明一下售货柜场景里消息内容本身不涉及用户隐私字段主要是设备ID、订单号、状态码这些因此应用层做了个简单的字段级签名校验避免伪造消息注入。2. MQTT主题与消息设计不丢单的核心功夫2.1 Topic规划三个层级定全局MQTT项目里第一个要紧的设计就是Topic。我在第一个版本里Topic起得乱七八糟设备上报、下发的路径混在一起后来排查线上问题找不到对应消息才狠下心来重新设计了整套主题体系。现在这个项目里所有Topic都遵循三条规范一是命名按“域/设备类别/设备标识/动作”四段式组织二是上行设备到服务端和下行服务端到设备分开三是不在Topic里放业务订单号这些随机字段订单相关信息一律放在消息体里。实际用到的核心主题如下vending/cmd/{deviceId}/open服务端下发开锁指令消息体包含订单号、超时时间vending/cmd/{deviceId}/close强制锁门管理后台远程操作vending/event/{deviceId}/state设备上报门状态、传感器数据这个是最高频的消息vending/event/{deviceId}/alarm设备上报异常告警比如门未关、重力异常vending/event/{deviceId}/heartbeat设备心跳包含电量、信号强度、固件版本为什么要按设备ID分Topic而不是所有设备共用一个Topic一开始我觉得共用一个Topic多省事服务端订阅一次就能收到所有设备的消息。但这样做的两个问题很快暴露出来一是没法利用MQTT本身的Topic权限做设备级隔离一台设备理论上能收到其他设备的信息二是调试的时候极其痛苦抓日志根本分不清消息来自哪台设备。按设备分Topic之后每个设备只能订阅自己的指令主题服务端也能按设备维度做消息追踪遇到问题一次就能定位到具体设备。2.2 QoS选型QoS 1是最优解MQTT有三档QoS我直接说结论这个项目里所有业务消息都用了QoS 1只有心跳用了QoS 0。为什么不用QoS 2QoS 2虽然能保证消息不重不漏但它需要四段握手每一条消息的确认开销很大用在售货柜这种高频状态上报场景下吞吐量上不去。实际业务里“重复”并不可怕我把消息处理做成了幂等——真正可怕的是“丢失”尤其是开锁指令丢了用户会被锁在柜子前面。QoS 1保证消息至少到达一次但会把重复消息的问题留给你。我在设备端和服务端都做了去重处理每一条业务消息都带一个自增的消息序号或业务单号接收方在Redis里维护一个最近N条消息ID的滑动去重集合重复的消息直接丢弃。Redis的SETNX配合过期时间特别好用比在内存里维护HashSet要可靠得多服务重启后去重状态也不会丢。关于Retain消息我只在设备启动时的初始状态同步里用了它。设备连接EMQX后向其状态Topic发布一条带Retain标志的当前状态快照这样即使服务端在设备离线期间错过了某些状态变化新上线的服务端订阅也能立刻拿到设备的最新状态不会出现“服务起来了但不知道设备什么状态”的尴尬。业务命令类消息一律不设Retain避免设备重连后收到一条几小时前的开锁指令那会引发严重的安全问题。2.3 遗嘱消息给设备“死亡”一个准确的信号在线状态判断是所有IoT项目都会遇到的问题。HTTP时代我们靠心跳超时判离线比如连续10个周期没收到心跳就认为设备挂了。但心跳超时有个问题服务端需要等待一段时间才能确认设备离线而且如果设备网络波动导致心跳延迟很容易误判。MQTT的遗嘱LWT机制从协议层面解决了这个问题——设备建立连接时带上遗嘱消息如果设备异常断开比如断电、断网Broker会立刻替设备把这条遗嘱消息发出去。我在设备上线时设置遗嘱内容为{online:false,deviceId:xxx,ts:...}发布到设备状态主题正常下线时设备会主动发一条online:false的消息并断开连接Broker不会再发遗嘱。这样服务端能在秒级内感知设备离线及时把设备标记为不可售。这里有个容易踩的坑设备SDK在设置遗嘱时一定要把遗嘱的Topic和正常上报状态的Topic保持一致否则服务端要订阅两个主题才能感知在线状态极容易漏消息。真实项目里出现过设备断网后App还在正常售货因为服务端没收到离线信号直到用户下单后开不了门才暴露问题。2.4 消息格式JSON不是性能瓶颈别过度设计消息体我用的是JSON而不是二进制或Protobuf。我知道很多IoT团队会强调二进制协议的各路优势但在这个项目里我坚定选择了JSON原因有三个第一售货柜消息体只有几十到几百字节JSON的解析开销完全可接受第二调试的时候mosquitto_sub订阅主题打印出来的消息直接可读不用写解码器第三硬件团队用的ESP32上有成熟的ArduinoJson库解析成本也在可控范围内。对于需要更低功耗的LoRa或NB-IoT场景我会推荐Protobuf但在4G网络环境下的售货柜JSON是投入产出比最高的选择。消息体我始终保持统一的信封格式{ msgId: uuid生成或者设备端递增id, deviceId: dv202406130001, type: DOOR_STATE, data: { door: OPEN, weight: 1.2, ts: 1718260000000 } }msgId是全链路幂等的关键字段服务端收到消息后先查这个Id是否处理过处理过直接返回。type字段用来标识消息类型避免一个Topic承载多种类型时的解析混乱。data里才是实际业务数据。这套信封格式从第一版沿用到现在后续扩展新消息类型时只增加type值服务端的事件分发逻辑完全不用改。3. SpringBoot服务端核心实现消息接入与业务落地3.1 依赖引入与核心配置SpringBoot整合MQTT我用了Spring Integration的spring-integration-mqtt它在MqttPahoClientFactory的基础上提供了消息驱动通道MqttInbound订阅到的消息会自动转成Spring的Message对象进入服务端管道。依赖引入如下dependency groupIdorg.springframework.integration/groupId artifactIdspring-integration-mqtt/artifactId /dependency dependency groupIdorg.eclipse.paho/groupId artifactIdorg.eclipse.paho.client.mqttv3/artifactId version1.2.5/version /dependencyapplication.yml里的关键配置项mqtt: broker: tcp://127.0.0.1:1883 client-id: vending-server-${random.value} username: vending password: xxx topic-pattern: vending/event/ server-qos: 1 keepalive: 60 connection-timeout: 10 max-inflight: 1000几个配置项的说明client-id必须唯一如果两个客户端用相同ID连接同一个BrokerEMQX会强制把旧连接踢掉这在生产环境会造成服务端反复断连keepalive我设60秒比设备端的120秒短这样Broker能及时感知服务端异常max-inflight控制未确认消息数量上限对QoS 1来说就是允许有多少条消息同时处于“已发送未确认”状态设备上报密集时这个值调大能减少阻塞。3.2 消息接入管道从MQTT到Spring事件Spring Integration把MQTT接入的方式是定义FlowMQTT入站适配器收到消息后交给处理器处理器解析信封并发布Spring事件。我在代码里把这件事做得尽量简洁清爽Bean public IntegrationFlow mqttInboundFlow( MqttPahoClientFactory factory, MessageHandler mqttMessageHandler) { return IntegrationFlows .from(new MqttPahoMessageDrivenChannelAdapter( vending-server, factory, vending/event/)) .handle(mqttMessageHandler) .get(); }关键的还是消息处理器的实现。这里我做了三件事解析Topic提取deviceId解析消息体做幂等校验然后按type字段分发到不同业务方法Override public void handleMessage(Message? message) { String topic (String) message.getHeaders().get( IntegrationMessageHeaderAccessor.HEADER_TOPIC); String deviceId parseDeviceId(topic); String payload (String) message.getPayload(); MqttEnvelope envelope JsonUtils.parse(payload); if (envelope null) return; if (!this.idempotentService.tryProcess(envelope.getMsgId())) return; switch (envelope.getType()) { case DOOR_STATE: this.deviceStateService.handleDoorState(deviceId, envelope.getData()); break; case ALARM: this.alarmService.handleAlarm(deviceId, envelope.getData()); break; case HEARTBEAT: this.deviceService.refreshHeartbeat(deviceId, envelope.getData()); break; default: log.warn(unknown message type: {}, envelope.getType()); } }IdempotentService是纯Redis实现setIfAbsent幂等标识过期时间设30秒——覆盖一条消息从入站到业务处理完成的耗时比时间短了可能会漏防重长了Redis内存备受压。批量开锁高峰期这个条件判断能兜住很大一部分重复消息吞吐量上有明显感知。3.3 订单中心的闭环设计“取货即走”的核心订单流程是用户扫码开门创建预订单 → 取货关门上报重量变化生成扣款订单 → 支付回调微信支付免密扣款 → 通知用户。我重点说说扣款金额怎么算。售货柜用的是重力传感器闭门后会上报一个重量值服务端拿这个重量减去初始货盘重量得到“减少的重量”然后匹配商品克重计算出实际拿走的商品数量和应付金额。这里有个细节同一种商品SKU的重量是固定的但传感器存在温漂和机械误差我做了2%的标准差容错。如果计算出来的数量有多解比如500g的结果可能是1瓶500ml的水也可能是2袋250ml的饮料系统会走人工介入流程——先把订单挂在待确认状态推送运营后台由人工确认扣款金额。这个兜底逻辑虽然看起来不“黑科技”但真实项目里没有它监控会接到大量投诉电话。订单状态机我用了最经典的待支付→支付中→已支付→已退款加上各种异常态。Redis贯穿整条链路预创建订单用Redis存了15分钟有效期的柜门令牌用户关门扣款时用Redis分布式锁防并发最后MySQL负责订单数据落库。简单说Redis管热数据MySQL管冷数据各自干擅长的事。4. 端到端业务流程从扫码开门到自动结算的全链路4.1 一次完整的“取货即走”时序下面是我整理的完整时序逻辑你对照自己项目的流程看看有没有落下关键环节用户在小程序扫柜门二维码小程序调服务端POST /api/v1/orders/prepare传设备ID和用户身份。服务端校验设备在线状态与售货状态生成预订单和15分钟有效的“开门令牌”通过MQTT向设备主题vending/cmd/{deviceId}/open下发开锁指令。设备收到指令后打开电磁锁向状态主题上报DOOR_STATE为OPEN服务端收到后更新预订单状态为“柜门已打开”开始计时。用户拿走商品关上柜门。设备检测到门锁到位等待重力传感器读数稳定后上报本次重量变化。服务端收到闭门消息此时订单中心开始算账重量差→商品匹配→金额确定。调用微信免密支付接口完成扣款。扣款成功订单状态置为“已支付”通过小程序订阅消息推送扣款明细扣款失败订单转待补缴锁定该用户后续补缴完成前无法再次使用。如果用户15分钟内未关门比如开了门啥都没拿就走了预订单自动取消服务端下发锁门指令并记录一次“超时未关”事件。这套流程最关键的体验点在于第5步和第6步之间的时间窗口。用户吃完饭关门就想看扣款结果结果服务端要等传感器数据稳定又要跑商品匹配还要调支付整个流程超过3秒就会出现大量客服咨询。我实测下来稳定状态下闭门到支付完成平均耗时1.8秒主要开销在传感器稳定周期约1秒和支付回调约0.5秒剩下的都是纯计算时间。4.2 超时、崩溃与异常兜底系统崩溃这件事在最坏的情况下会发生用户关了门服务端掉电了传感器数据没来得及处理。订单就这么挂在“门已打开”状态用户也走了柜门还锁着商品少了钱没扣到。针对这种情况我设计了三个兜底机制。第一个是“状态补偿扫描”定时任务每5分钟扫描一次“门已打开”状态超过10分钟的预订单主动向设备查询当前闭门状态和重量读数如果设备已关门则补跑结算流程。第二个是“异常订单看板”所有超过3分钟未结算到已支付状态的订单实时推送到运营后台人工可以一键介入补单或发起退款流程。第三个是“断电恢复补报”设备端固件上有本地断电续传逻辑设备重启后会把未上报的门状态和传感器记录补发一遍兜住设备断电后丢消息的最极端场景。这些兜底机制单独拎出来都很简单但组合起来才能形成一个基本闭环的无人零售系统。我给项目做故障注入测试的时候发现总体丢单率从第一版的5%降到了稳定运行后的0.1%以下这0.1%基本都发生在设备侧传感器彻底损坏的物理故障场景这部分只能靠定期巡检去覆盖了。4.3 WebSocket推送让用户实时看到扣款结果支付结果要通过小程序通知用户我选择了WebSocket实时推送而不是传统的小程序订阅消息。原因很简单用户关门的瞬间人还在柜子面前小程序页面正停留在订单详情页WebSocket推过来的扣款提醒直接刷新当前页面体验上是“关了门就看到钱扣了多少”而不是等一个系统通知。我在SpringBoot里用WebSocketHandler维护了一个userId → Session的映射表把扣款结果通过JSON直接推给前端。设备上报→结算→WebSocket推送整条链路在500ms内完成用户体感几乎是实时的。WebSocket只在用户打开小程序页面时建立页面关闭连接就断开所以对服务端的长连接压力在可控范围内。测试下来单机几千并发在线没有明显性能问题。如果后续用户量上来了横向加机器部署WebSocket集群时要注意Session共享问题那又是个新故事了本文先不展开。5. 典型故障与排查技巧实录5.1 消息重复导致重复扣款上线第一周就翻过车用户拿了一瓶水扣了两次款。排查发现是网关设备在信号弱的时候把QT消息重发了一次设备端没有处理好去重逻辑服务端收到的msgId相同但我的幂等服务竟然没拦住。为什么会没拦住因为我把幂等过期时间设成了30秒而网关重发消息时已经过去了40秒。在弱网环境下MQTT QoS 1的重发可能延迟很久远大于30秒的窗口。这个问题给了我一个教训幂等窗口不能拍脑袋定要看业务链路的最大允许延迟。最后我把幂等过期时间调整为10分钟同时把订单支付环节也做了数据库层的唯一约束——一个订单只能有一条成功支付流水。两层防线再也没出现过重复扣款的投诉。5.2 断线重连风暴打爆Broker第二批机器上线的时候出现过一次EMQX CPU打满的情况。排查后发现是运营商在凌晨4点做网络割接所有4G设备同时断开重启后全部在极短时间内尝试重连EMQX承受不住几千个连接同时建立的握手压力直接假死殃及所有在线设备。解决办法有两层。第一层是在EMQX配置里打开了zone.max_connections限流和自动背压第二层更关键在设备端固件里做了“重连退避策略”——第一次断线立即重连重连失败后等待随机2到5秒之后每次等待时间按指数增长最多不超过5分钟并且每次重传前加上设备ID的哈希散列让所有设备错峰连接。这个策略上线后即便是大规模断网复网EMQX的CPU峰值也稳稳控制在30%以内。5.3 时序错乱先收到闭门后收到开门MQTT的QoS保证的是单条消息的送达不保证消息顺序。实际运行中我遇到过一种诡异情况设备上报DOOR_CLOSE的消息竟然比DOOR_OPEN先到达服务端。虽然发生概率极低但一旦发生订单中心就会拿错误的重量数据去扣款。处理这个问题的思路是“服务端不做顺序假设用状态机校验”。设备上报状态变化时服务端检查当前存储的设备状态如果从“已关门”直接跳变到“已关门”这个变化和已知状态矛盾判定消息乱序丢弃无效消息并标记一个设备侧异常等待下一次心跳状态对齐。同时设备端在固件里给状态消息加了递增序号服务端发现有消息序号断层时主动发起状态同步请求而不是盲目处理乱序消息。5.4 用MQTT Explorer做日常排障的效率飙升处理MQTT项目强烈建议装一个MQTT Explorer做日常调试和排障。它能可视化订阅多个主题实时看到每个主题的消息流还能用类似SQL的过滤器筛选特定设备ID的消息。每次运营反馈“某台设备异常无操作”我都能通过MQTT Explorer订阅该设备的所有主题查看最近的上报记录和指令下发记录在几分钟内判断是设备掉线、指令没下发还是服务端处理出错不需要翻一堆分散的日志。我日常排障的固定操作流程是这样的MQTT Explorer订阅vending/event/#和vending/cmd/#过滤目标设备ID查看设备有没有上报心跳没有则先确认设备和网络状态有上报但服务端没回应看服务端日志里有没有报错或幂等拦截定位到服务端处理的断点后再结合Redis里的订单状态判断具体卡在哪个环节。这套流程在四个故障案例里都帮我快速定位了问题从原来的半小时排查时间缩短到10分钟以内。5.5 性能与容量一台服务器能扛多少台设备最后聊一个运营经常问的问题这套系统一台服务器能扛多少台设备我以一台4核8G的云服务器、EMQX和SpringBoot同机部署为例给出一个实测参考数据EMQX侧稳定支撑3000台设备的连接和消息转发没有问题这时SpringBoot的处理速度会成为瓶颈瓶颈的主要来源是订单结算时对Redis和MySQL的访问开销。如果每台设备平均每30秒上报一条状态消息一秒约100条消息进入业务处理4核机器CPU跑在70%左右。要扩容时优先把EMQX单独拆出去部署再给MySQL加从库分担读流量这样把单机支撑量提升到5000台以上没有太大压力。关于SpringBoot的线程池这里有一个值得注意的经验MQTT入站的消息处理是阻塞的如果把消息接入直接跑在Tomcat的工作线程里一旦业务处理中有数据库慢查询整个消息通道都会被拖慢。标准的做法是给消息处理单独配一个线程池控制并发数并且使用独立的数据库连接池资源避免消息处理把请求线程池打满影响REST接口的响应。我用了ThreadPoolTaskExecutor核心线程数设8、最大16、队列容量2000实测效果良好。6. 给想做类似项目的朋友几句真心话这套系统从需求确认到第一期落地前后大约三个月。回头看技术选的并不复杂——SpringBoot和MQTT都是各自领域里非常成熟的东西真正花时间的是边界场景的打磨弱网重传、异常状态恢复、重复消息兜底、金额计算容错没有一个是可以跳过的“锦上添花”。奉劝想快速复刻一个“取货即走”项目的朋友把60%的时间留在业务闭环和异常处理上而不是一味追求炫酷的设备端特效和新框架的酷炫用法。几个关键建议再重复一遍Topic规划必须一开始做全局设计不要等设备量大了再回头改一切设备消息必须带上全局唯一ID这是后续所有幂等和排查工作的前提流程设计上永远要考虑服务端掉电、设备断网、用户乱操作这三类“故障场景”。把这三件事想透你的项目大概率能比市面上很多售货方案稳得多。
