物联网平台同时支持TCP和MQTT接入：架构设计与实战经验

做物联网平台的接入层绕不开两个协议TCP和MQTT。最近我正好完成了一个同时支持这两种协议的物联网平台从架构设计、协议接入到设备管理全链路跑通过程中踩了不少坑也沉淀了一些可以直接复用的经验。这篇文章就把完整思路和关键实现拆开讲清楚帮正在做类似平台的你少走弯路。这个平台面向的场景很典型一端是大量通过TCP长连接上送的工业网关、DTU、嵌入式设备另一端是需要实时订阅设备状态的业务系统、小程序、监控大屏。TCP方式灵活但协议格式需要自己定义和解析MQTT自带发布订阅模型和QoS等级对弱网设备更友好。平台把两者统一收拢为“设备接入层”向上提供一致的设备数据接口业务侧不需要关心设备到底是用TCP还是MQTT上来的整体设计就清爽了。1. 项目背景与总体设计思路1.1 为什么同时支持TCP和MQTT先看接入端到底有什么很多做物联网平台的人第一反应是“直接用MQTT不就行了”但真去现场看过设备接入情况就会发现事情没那么简单。市面上现存设备里基于TCP私有协议接入的存量设备非常多尤其是工业场景——PLC通过Modbus TCP网关把寄存器数据发到服务器环保监测设备按HJ 212协议走TCP上报充电桩、光伏逆变器、消防主机每一类设备都有自己的一套报文规范这些报文基本都是定长或变长的十六进制帧靠TCP长连接维持。如果平台只做MQTT这些存量设备就全都要改固件、改协议根本不现实。而且有些场景下TCP就是比MQTT更合适设备侧就是个裸的单片机系统内存十几KB跑MQTT库要占用大量资源还有一些私有通信是边采集边推不需要订阅模型TCP直连反而链路最短、延迟最低。反过来对业务侧来说MQTT的价值又非常明显。同一份设备数据可能要同时推给多个业务方智能路灯的状态要同步到运营后台、告警也要推给值班小程序如果每条数据都走TCP点对点去分发连接管理就是一场灾难。MQTT的发布订阅模型天然解决多订阅者分发问题而且QoS机制能保证消息不丢。所以这个平台从一开始就定了原则接入层兼容多协议业务层统一模型。TCP和MQTT不是二选一而是互补。1.2 平台整体架构与模块划分整个平台按功能拆成五层层与层之间通过内部接口通信互不感知协议差异。第一层是接入层负责维持设备长连接、解析原始报文、处理心跳、管理连接生命周期。TCP接入服务用Netty实现每个设备连接对应一个Channel通过自定义解码器把字节流解析成标准设备消息MQTT接入服务直接基于EMQX Broker平台侧通过订阅主题和Webhook接收设备事件。第二层是会话管理层维护设备与平台之间的映射关系。设备上线后平台为它建立Session记录设备信息、连接标识、最后活跃时间设备离线时Session释放并触发离线事件。TCP和MQTT两种接入方式最终都归一化成统一的Session对象。第三层是消息路由层负责把设备上送的数据根据规则转发到对应的业务服务。它网关注册了设备类型的处理器比如采集数据、告警上报、远程下发、OTA进度等每条消息按类型路由到对应的业务模块。第四层是业务服务层包含实时监控、设备影子、告警管理、数据存储这些核心功能。设备影子保存设备的最新状态业务系统读影子就相当于读设备快照不用反复询问设备。第五层是存储层关系数据库存设备档案、用户、告警记录时序数据库存设备遥测数据Redis缓存在线状态和会话信息。我在实际搭建过程中最大的体会是协议接入做得再花哨如果会话管理层不做好抽象后面每接一种新协议就要改一堆业务代码。所以设备信息模型和数据上报格式一定要在协议接入层之前定义清楚。1.3 技术选型为什么是Netty EMQX Spring Boot技术选型这块原则不追求新只追求稳。TCP接入服务选择了Netty理由是它解决了Java NIO里最复杂的线程模型和ByteBuf内存管理问题Netty的pipeline机制可以非常优雅地组合解码器、心跳处理器、业务分发器。而且Netty的社区活跃度高生产环境验证充分不用担心踩到没人填过的坑。MQTT Broker直接选了EMQX这个决定帮我们省了至少一个月的工作量。MQTT Broker看起来只是个消息中间件但真正做起来要考虑集群、持久化、认证鉴权、主题权限管理、规则引擎、监控告警这些EMQX都是开箱即用的。如果要自己用Moquette或者Netty实现MQTT协议光是把QoS2的报文交互流程测试完全工作量和踩坑成本都非常大。后端业务服务用Spring Boot这是Java生态里最主流的选择生态完整接数据库、Redis、消息队列都是现成的starter。设备接入和业务服务之间用RocketMQ解耦接入层只负责收数据、发数据不做任何业务处理这样接入层可以横向扩容设备连接被负载均衡分散到多台接入节点。这个组合下来整套平台的维护成本被压得很低。我见过一些团队为了炫技用Go写接入层用Node写后台用Python写算法服务最后运维光维护语言环境就苦不堪言。做平台类项目工程化成熟度比技术新颖度重要得多。2. 协议核心概念与接入链路设计2.1 从TCP三次握手说起连接这件事没那么简单做TCP接入服务首先得把连接建立这块理解透。TCP是面向连接的可靠传输协议通信双方通信前需要先建立连接这个过程就是著名的三次握手。客户端发SYN报文请求建立连接服务端回SYNACK表示收到并同意客户端再回ACK确认连接就算建立了。三次握手的价值在于它同时确认了双方的收发能力都正常也交换了初始序号为后续可靠传输打下基础。我在调试设备接入时经常干一件事就是在服务端用抓包工具看SYN报文有没有正常进来如果设备不断重发SYN但服务端没有回包基本可以断定是端口没监听成功或者防火墙把包丢了。连接建立之后TCP还负责流量控制和拥塞控制。设备端因为网络波动导致大量丢包重传时服务端可能会观察到持续的窗口缩小通告这时候设备的上报延迟会明显增加。在物联网场景里这个现象经常被误判为设备卡死实际上是网络拥塞导致TCP在自适应降速耐心等网络恢复就会自动恢复。这里我给一个排查建议当设备反馈“有时连得上有时连不上”时除了看应用日志还要在服务端抓包看SYN重传。SYN重传次数超过阈值说明客户端到服务端的路径上有丢包再往后排查交换机、Wi-Fi环境、SIM卡信号强弱方向会更清晰。2.2 MQTT的发布订阅模型与QoS等级设计MQTT是构建在TCP之上的应用层协议虽然底层也是TCP连接但它把通信模型从“点对点”改成了“发布订阅”。设备不再是直接连平台服务器收发数据而是连到一个Broker往某个主题发布消息需要这份数据的系统订阅这个主题就能持续收到数据。这个模型最大的好处是解耦。设备不知道数据被谁消费订阅方也不知道数据从哪个设备来。平台侧要新增一个数据消费者只需要增加一个订阅设备侧零改动。业务告警、实时大屏、移动端推送、数据入库这些消费者都是挂到主题下自动获得消息的。MQTT的QoS服务质量是接入设计里最容易让人纠结的点。QoS0最多一次消息发了就不管效率最高但有丢失可能QoS1至少一次Broker收到消息后回PUBACK如果客户端没收到PUBACK会重发可能产生重复消息QoS2恰好一次通过四步握手确认最可靠但性能和实现复杂度最高。在实际工程项目里QoS2基本被淘汰了因为绝大多数场景QoS1配合幂等处理就能满足要求而且QoS1的语义对设备端友好设备只需要存一条重发队列就行。我在平台里对所有上行数据都设计为QoS1业务侧在消费时按照消息ID做去重把“至少一次”变成“恰好一次”成本和效果都最优。设备下行控制指令同样用QoS1但要求设备侧实现指令幂等同一条指令重复执行也不产生副作用这样网络抖动重发就不会造成开关反复动作之类的设备安全事故。2.3 接入链路参数设计心跳、端口、超时、报文大小接入链路上最容易忽视但又极其重要的四个参数是心跳、端口、超时和报文大小。TCP接入的心跳机制我推荐的是由平台侧主动探测为主、设备侧主动上报为辅的双向策略。设备每30秒上报一次应用层心跳应用层心跳的好处是能顺带携带设备状态如果设备来不及发心跳服务端每90秒检测一次空闲没有收到任何报文就主动断开这条僵死连接。设备侧探测平台是否存活则通过设置Socket读超时为120秒实现120秒内平台不推送任何数据就尝试重连。这里特别提醒一个细节不要指望TCP的keepalive机制来维持连接。TCP keepalive默认要7200秒无数据才触发探测而且触发完全由内核控制应用层拿不到可靠的通知对物联网设备的连接保活来说基本没有帮助。应用层心跳才是正解。MQTT连接的话心跳由Keep Alive字段控制平台侧建议设置为60秒设备在60秒内发送任意报文包括PINGREQ就算活跃。EMQX默认允许客户端心跳周期在一定范围内浮动建议关掉这个浮动限制强制所有设备保持一致的心跳周期避免某些以“省电”为由把心跳调到很长的设备占用大量空闲连接。端口设计上TCP接入服务默认监听7001端口用于设备接入EMQX监听1883端口作为MQTT接入端口。生产环境里这两个端口要提前在防火墙、安全组都配置放行并在接入层服务器上用iptables做源IP和端口限流。我遇到过一次端口忘放行导致设备批量连接失败的事故排查了大半天才发现是安全组规则没来得及更新。报文大小也要限制。TCP服务端通过配置Netty的MaxFrameLength限制单包最大长度MQTT则在EMQX配置max_packet_size超限直接断开。这个限制主要是防止脏数据和恶意攻击把内存耗尽。平台里先把最大报文长度设为32KB后续根据具体设备的报文类型再针对性调整。3. 核心模块实现与代码解析3.1 TCP接入层用Netty实现可靠的数据接收TCP接入层是整个平台里最核心的代码模块它要处理的核心问题有三个拆包粘包、心跳检测和业务分发。我先给出Netty的pipeline配置代码然后逐个说明。Configuration public class TcpServerConfig { Bean public ServerBootstrap serverBootstrap() { EventLoopGroup bossGroup new NioEventLoopGroup(1); EventLoopGroup workerGroup new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2); ServerBootstrap bootstrap new ServerBootstrap(); bootstrap.group(bossGroup, workerGroup) .channel(NioServerSocketChannel.class) .option(ChannelOption.SO_BACKLOG, 1024) .option(ChannelOption.SO_REUSEADDR, true) .childOption(ChannelOption.TCP_NODELAY, true) .childOption(ChannelOption.SO_KEEPALIVE, true) .childHandler(new ChannelInitializerSocketChannel() { Override protected void initChannel(SocketChannel ch) { ch.pipeline() .addLast(new LengthFieldBasedFrameDecoder(32 * 1024, 4, 2, -6, 0)) .addLast(new DeviceMessageDecoder()) .addLast(new IdleStateHandler(90, 0, 0, TimeUnit.SECONDS)) .addLast(new HeartbeatHandler()) .addLast(new DeviceMessageDispatcher()); } }); try { bootstrap.bind(7001).sync(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } return bootstrap; } }拆包粘包是TCP字节流模式带来的必然问题。TCP是流式协议没有消息边界应用层发100个字节接收方可能一次收到100字节也可能分3次收到40、20、40字节这叫拆包反过来多个小包粘在一起到达这叫粘包。解决办法就是要约定一条消息的长度边界。上门的设备协议格式做了统一约定完整帧包含4字节魔数0x5A5A5A5A、2字节总长度字段、2字节类型字段、N字节业务数据。LengthFieldBasedFrameDecoder的参数含义很清晰最大帧长度32KB长度字段偏移4字节长度字段本身2字节长度调整值-6是因为总长度包含了帧头6字节这样解码器裁掉长度字段后剩下的0字节头就是完整的业务数据包。这里很容易搞错的一点是LengthAdjustment的计算如果长度字段定义的是“从长度字段之后到帧尾的长度”那长度字段本身占用字节数也要算进Adjustment里否则取出来的数据总会多出几个字节解析必然报错。解码器把字节流转成统一的DeviceMessage对象。设备ID的提取逻辑是解析出帧内容后约定帧内第一个字段是设备ID字符串如果首包中设备ID为空或非法直接关闭连接。这个设计把设备标识和设备连接绑定起来后续所有报文都带上这个设备ID进入业务分发。心跳处理用Netty的IdleStateHandler实现90秒空闲就触发一个事件由HeartbeatHandler统一断开连接并清理Session然后发送设备离线事件到消息队列。这段代码里有个细节值得说下IdleStateHandler要放在解码器之后因为心跳判断的是“应用层有没有数据进来”如果放在解码器前只要底层TCP有数据就算活跃但那些数据可能解码失败设备实际已经断了照样测不出来。3.2 MQTT接入层搭建Broker并打通消息管道MQTT接入层我没有自己写Broker而是用EMQX做主接入服务平台业务服务通过Spring Boot订阅EMQX上的主题拿到设备数据再投递给后面的业务处理器。这样做的好处是接入并发能力直接交给EMQX去扛平台侧代码只需要消费消息。EMQX安装完成后需要检查几个关键配置项。第一是认证配置启用密码认证并在EMQX Dashboard里创建应用和设备账号。第二是Topic权限控制设备只能发布指定的数据主题只能订阅平台下发的指令主题通过ACL规则限制避免设备间越权互访。第三是打开Webhook或规则引擎把设备上下线事件、消息投递事件转发到平台的HTTP接口平台侧统一收口事件流。# application.yml 中MQTT配置片段 mqtt: broker: url: tcp://127.0.0.1:1883 username: iot_platform password: iot_platform_pwd client-id: platform-service topic: device-data: /device/data/# device-event: /device/event/# device-command: /device/command/#Spring Boot工程里我用的是Eclipse Paho的Java客户端声明一个MqttClient连接EMQX同时订阅两个主题设备数据主题和设备事件主题。Component public class MqttConsumer { Value(${mqtt.broker.url}) private String brokerUrl; Value(${mqtt.broker.username}) private String username; Value(${mqtt.broker.password}) private String password; Value(${mqtt.broker.client-id}) private String clientId; private MqttClient client; PostConstruct public void init() throws MqttException { client new MqttClient(brokerUrl, clientId, new MemoryPersistence()); MqttConnectOptions options new MqttConnectOptions(); options.setUserName(username); options.setPassword(password.toCharArray()); options.setAutomaticReconnect(true); options.setCleanSession(true); options.setConnectionTimeout(10); options.setKeepAliveInterval(60); client.setCallback(new MqttCallback() { Override public void connectionLost(Throwable cause) { System.err.println(MQTT连接断开等待自动重连); } Override public void messageArrived(String topic, MqttMessage message) { MessageRouter.route(topic, new String(message.getPayload())); } Override public void deliveryComplete(IMqttDeliveryToken token) { } }); client.connect(options); client.subscribe(new String[]{deviceDataTopic, deviceEventTopic}, new int[]{1, 1}); } }这里有几个容易踩的坑。第一个是clientId必须保持唯一如果多个服务实例用同一个clientId连接EMQX后连的会把先连的踢下线平台扩容时必然出事故。生产环境里我按实例名拼接clientId比如platform-service-node1。第二个是自动重连打开后重连成功要重新订阅一次主题Paho的自动重连不会自己恢复订阅。第三个是设备数据主题的QoS设为1消息到达可能重复业务侧必须按消息ID做去重。设备上送的MQTT payload格式与TCP接入层解析出的DeviceMessage保持一致都是JSON格式包含设备ID、消息类型、数据体、时间戳。这样到业务层就不用区分这个消息是TCP设备来的还是MQTT设备来的。3.3 设备影子和上下线管理设备影子是我在这个平台里特意加的一个模块它的作用像给每个物理设备配了一个“云端替身”业务系统读写影子不直接接触设备降低耦合也提升稳定性。影子数据结构包含四部分desired期望状态、reported实际状态、online在线状态、lastUpdate最后更新时间。设备上报状态后平台更新reported字段业务系统下发指令时先写desired字段然后下发到设备设备确认执行后再上报最新状态平台用reported覆盖desired。上下线管理依赖接入层上报的事件。TCP接入层在连接建立、断开时分别产生DEVICE_ONLINE和DEVICE_OFFLINE事件写进Redis的在线状态字段里过期时间设为心跳间隔的3倍MQTT接入层则通过EMQX的客户端连接事件和断开事件转发到业务侧统一触发同样的上下线逻辑。这样统一的处理方式能保证不管设备通过哪种协议接入业务侧看到的上下线状态语义完全一致。Service public class DeviceSessionService { Autowired private StringRedisTemplate redisTemplate; private static final String ONLINE_KEY_PREFIX device:online:; public void online(String deviceId, String protocolType, String gatewayId) { String sessionKey ONLINE_KEY_PREFIX deviceId; MapString, String sessionInfo new HashMap(); sessionInfo.put(protocolType, protocolType); sessionInfo.put(gatewayId, gatewayId); sessionInfo.put(onlineTime, String.valueOf(System.currentTimeMillis())); redisTemplate.opsForHash().putAll(sessionKey, sessionInfo); redisTemplate.expire(sessionKey, 3 * 60, TimeUnit.SECONDS); } public void heartbeat(String deviceId) { String sessionKey ONLINE_KEY_PREFIX deviceId; if (Boolean.TRUE.equals(redisTemplate.hasKey(sessionKey))) { redisTemplate.expire(sessionKey, 3 * 60, TimeUnit.SECONDS); } } public void offline(String deviceId) { String sessionKey ONLINE_KEY_PREFIX deviceId; redisTemplate.delete(sessionKey); } }这里用Redis的过期时间实现“软在线”机制是为了避免极端情况下连接异常断开导致离线事件丢失。只要设备持续上报心跳就会不断刷新过期时间一旦停止状态自动过期即使离线事件没有正常发出业务侧查询状态也不会误判设备在线。4. 实操中的坑与排查实录4.1 粘包/拆包是最常见的TCP接入问题接入层上线第一天就遇到设备数据大量解析失败看日志全是“invalid frame length”的报错。用抓包工具看实际TCP流发现设备端每100ms采集一次数据如果采集周期过短多个采集报文几乎同时到达服务端Netty的ByteBuf里一次就堆积了三四帧数据解码器按单帧长度解析自然错位。解决思路很清晰拆包问题用解码器解决粘包问题也要靠解码器解决。我用的是LengthFieldBasedFrameDecoder先读4字节魔数确认帧头再取2字节长度字段确定本帧总长Netty会按长度自动切分出完整帧粘包里的每一帧都能被正确剥离拆包后不完整的帧会先缓存等剩余字节到达再组包。这个过程透明完成业务代码完全不用感知TCP的流式特征。但这里有个容易忽略的细节长度字段的值不能只算业务数据必须包含除了长度字段本身之外的全部字节也就是帧头业务数据。如果设备端定义长度字段时只算了业务数据长度解码端的LengthAdjustment配置就要相应调整。4.2 端口绑定失败与连接重置问题本地调试TCP接入服务时遇到过启动直接报“error: listen tcp 127.0.0.1:7001: bind: only one usage of each socket address”这个错误。这个报错的意思很直白7001端口已经被别的进程占用了操作系统不允许同一个IP和端口被绑定两次。排查步骤如下先用netstat -ano | findstr 7001命令查看是哪个进程占用了端口再用tasklist /fi pid eq 1234定位进程名如果确定是无用的僵尸进程直接结束它再重启服务。如果端口被正常业务占用就要改配置换端口。这种情况其实很常见尤其是多人协作开发时有人在本机起了一个实例没关另一个人启动服务就撞上了。还有一个排查经验分享给读者线上环境里“TCP connection reset by peer”这个报错很多情况不是代码问题而是对端设备主动连接了但握手没完成就断开了或者中间网络设备的空闲连接回收策略把空闲连接重置了。比如运营商NAT设备通常只保留数分钟的空闲会话设备长时间不发数据NAT表项会被回收连接就被悄悄断开。遇到这种场景除了缩短应用层心跳周期没有更好的办法。4.3 MQTT客户端连上就掉线的排查用MQTTX调试设备连接时发生过几次“刚连接就掉线”的问题而且在局域网内测试正常到了4G网络环境就频繁掉线。当时排查了EMQX日志、看客户端返回码最后定位到是Keep Alive设置和客户端超时处理不匹配导致的。MQTT协议连接报文里带Keep Alive字段单位是秒。Broker如果在该时间周期的1.5倍内没收到客户端任何报文就会主动断开连接。有些IoT SDK默认把Keep Alive设置为0意思是永不过期或者设置得很大看起来连上了实际上一旦网络出现短暂波动TCP链路已经断了SDK却要等到很久才能感知期间的状态是假的“在线”。后来我把设备侧Keep Alive统一调整为60秒并且让SDK的心跳发送间隔为Keep Alive的1/3也就是20秒一个PINGREQ。这样Broker侧能及时感知心跳失败设备侧也能快速发现连接异常并重连实测在弱网环境下连接稳定性明显提升。还有一类掉线问题是clientId重复。两台设备共用一个clientIdMQTT协议规定新连接会踢掉旧连接两台设备就会交替掉线。排查时在EMQX的Dashboard里看连接列表如果同一个clientId反复上下线基本可以确定是分配clientId的逻辑出问题了。4.4 平台显示离线但设备还在上报这个问题是设备上送正常但平台侧在线状态长期显示离线。查下来发现是TCP接入层把设备解析出的设备ID和MQTT设备的设备ID没统一——TCP设备解析出来的ID带前后空格存进Redis的Key就成了“device:online:1001”但业务侧查询用的是“device:online:1001”加上固定前缀拼错了。后来把设备ID的标准化工作统一放到接入层完成所有协议接入后的设备ID都经过trim、大小写转换、统一去空格的处理再进入会话管理。这个教训很有价值——多协议平台最怕的其实就是不同协议的接入路径各自为政设备和连接的管理必须收敛到一个统一的标准入口上。另外一个隐蔽原因是设备影子缓存了旧的在线状态。Redis里的过期键虽然自动删除但查询时如果用exists命令判断过期键在键TTL还没完全走完前会得到true。所以在线状态查询我建议显式比较最后心跳时间与当前时间的差值超过阈值直接判定离线不依赖Redis的过期机制。4.5 常见问题速查表问题现象可能原因排查与解决TCP客户端连接失败SYN无响应端口未监听、防火墙拦截检查服务监听状态尝试关闭防火墙测试连接建立后频繁断开应用层心跳发送间隔超过服务端超时客户端心跳间隔 ≤ 服务端空闲超时的一半报文解析出现乱码或错位粘包/拆包、长度字段配置错误用LengthFieldBasedFrameDecoder核对LengthAdjustmentMQTT客户端连接被拒绝用户名密码错误、ACL未放行、clientId冲突检查EMQX Dashboard认证记录与连接日志MQTT连上后又断开Keep Alive过长或为0、clientId重复统一Keep Alive为60秒保证clientId唯一设备显示离线但网络正常在线状态查询逻辑不统一、缓存过期机制异常统一设备ID标准化处理显示比较最后心跳时间消息重复处理MQTT QoS1语义导致重复消息业务侧按消息ID做去重保证处理幂等5. 复盘与个人体会这个平台从搭骨架到跑通全链路前后花了不到三周时间核心收获不在代码量而在对“物联网接入”这件事的整体认知。第一协议兼容的核心不是把每种协议都实现一遍而是定义一个统一的设备信息和消息模型让TCP接入、MQTT接入都收敛到同一套标准上。后面再要接入HTTP协议设备、LWM2M设备只需要新增一个协议适配层业务完全不受影响。第二连接管理和业务处理必须彻底分离。TCP通道、MQTT会话这些是接入层的事情数据解析、消息路由、设备影子是平台层的事情两者不要混在一起。否则一旦设备量上来接了新协议就改业务日子会非常难过。第三所有验证过可行的参数都要固化成文档或配置模板不要靠口头传。心跳间隔、端口规划、报文长度限制、clientId生成规则、超时时间这些参数看起来不起眼实际上全是线上事故的源头。后面这个平台还可以继续扩展比如接入LwM2M协议支持NB-IoT设备把EMQX的规则引擎用起来做数据清洗甚至对接TSDB做设备时序数据的长期存储分析。核心架构只要保持接入层与业务层解耦这些扩展都只是增加适配器的工作量。如果你也在做类似的物联网平台建议先把TCP和MQTT这条主链路跑通它就是整个平台的地基地基好了一切都顺。