前面的文章把设备怎么连上来一机一密 ACL、断线了怎么办自动重连、遗嘱消息都讲过了。但有一条主线一直没完整写过一条遥测消息从 EMQX 到达服务端之后、落进数据库之前到底经历了什么这条链路是物联网后端的主干道。面试里说说你的平台怎么处理设备消息出现频率极高因为这一个 problem 就能问出你是不是真做过Topic 怎么设计的、消息怎么校验、脏数据怎么办、为什么这么分层。这篇就把我的真实代码摊开讲。一、先约定 Topic它就是设备端的接口契约我平台的消息处理入口是TelemetryMessageHandler.java文件头注释里写着整个系统的主题约定/** * 主题约定 * up/{productKey}/{deviceId} 遥测数据payload 为 JSON * st/{productKey}/{deviceId} 设备状态payload 为 online/offline遗嘱消息用 offline */只有两条主题、三层路径。为什么设计得这么抠因为 Topic 在物联网里就是接口契约——它是设备端和后端共同遵守的 URL定下来就不能随便改。层级越少设备端固件越好写后端解析越不容易出错通配符订阅也越方便服务端只要订阅up/#和st/#两条就能收全所有产品的所有设备。做后端的可以这么类比Topic 是 URL pathpayload 是 request bodyQoS 是要不要重试。把 REST API 设计里契约先行的纪律原样搬过来就行——先定 Topic 和 payload 格式再写解析代码顺序不能反。二、回调函数里第一件事别让一条消息弄死连接消息从 EMQX 推到服务端进入的是 Paho 客户端的回调messageArrived。我在MqttConnection.java里是这么写的OverridepublicvoidmessageArrived(Stringtopic,MqttMessagemessage){StringpayloadnewString(message.getPayload(),StandardCharsets.UTF_8);try{handler.handle(topic,payload);}catch(Exceptione){// 单条消息处理失败不能影响连接和后续消息log.error(处理消息失败: topic{},topic,e);}}重点是这个 try-catch。messageArrived跑在 Paho 客户端单条的回调线程上业务代码在这里抛异常后果不只是丢一条消息——处理流程卡住后续消息全部排队甚至影响心跳和连接稳定性。所以回调里只做两件事转码、委托给业务 handler并且任何异常都必须就地吞掉记日志。连接参数上我也做了取舍这几行在面试里被追问的概率很高options.setAutomaticReconnect(true);options.setCleanSession(true);options.setMaxInflight(100);// 订阅时int[]qos{1,1};QoS 用 1遥测数据允许重复但不允许丢cleanSession 用 true服务端掉线期间的旧消息不强求补齐因为看板要的是当前状态几秒前的旧温湿度补回来反而是脏数据。这是业务属性决定的选型不是拍脑袋。三、三层拦截认证、鉴权、业务校验各拦各的真正进业务逻辑之前每条消息要过三道门。第一道在 EMQX一机一密认证 ACL前面有文章专门写过但后端代码里还有两道DevicedevicedeviceRepository.findById(deviceId).orElse(null);if(devicenull||!productKey.equals(device.getProductKey())){log.warn(收到未知设备的数据已丢弃: topic{},topic);return;}// 已禁用的设备即使 Broker 层有漏网的ACL 未生效/缓存入库前再拦一道。// 安全上的原则认证、鉴权、业务校验三层各拦各的不假设上一层一定生效。if(DeviceService.STATUS_DISABLED.equals(device.getStatus())){log.warn(设备[{}]已禁用丢弃其消息: topic{},deviceId,topic);return;}为什么要重复拦截因为每一层防的风险不一样**第一层认证**防你是谁都没搞清——伪造 clientId、偷来的密钥**第二层ACL**防合法设备发不该发的主题——但 ACL 有缓存、有生效延迟**第三层业务校验**兜住前两层的一切漏网被禁用的设备、topic 里 deviceId 和 productKey 对不上的伪造请求。这段代码注释里那句话是我踩坑之后的总结“不假设上一层一定生效”。分布式系统里每一层都可能有缓存、延迟、配置漂移把安全押在前面那层应该拦住了上迟早出事故。四、先存原始 JSON再谈解析过了三道门的遥测消息我的处理方式可能和很多教程不一样——不做字段级解析原始 payload 直接落库privatevoidhandleTelemetry(Devicedevice,StringproductKey,Stringpayload){// 校验是合法 JSON不合法直接丢弃防脏数据进库try{objectMapper.readTree(payload);}catch(Exceptione){log.warn(payload 不是合法 JSON丢弃: {},payload);return;}device.setStatus(online);deviceRepository.save(device);DataPointdataPointnewDataPoint();dataPoint.setDeviceId(device.getId());dataPoint.setProductKey(productKey);dataPoint.setPayload(payload);// 原样保存不拆字段dataPoint.setReceivedAt(Instant.now());dataPointRepository.save(dataPoint);}两个细节。第一入库前用readTree做一次轻量校验只确认这是合法 JSON不提取任何字段——防的是固件 bug 吐出来的半截数据污染表。第二payload字段存的是原始字符串收到时间单独记一列。为什么这么设计因为设备端是最不可控的一环固件有 bug、传感器会吐异常值、协议后面还会升级。解析逻辑一定会变但原始数据错了就永远错了。只要原始 JSON 在以后想加字段、改解析规则、修历史数据重放一遍就行。做数据平台的人都懂一句话最值钱的不是解析后的结果是带时间戳的原始数据。另外注意device.setStatus(online)这行——收到数据本身就是最可靠的在线信号这比心跳轮询更及时离线判定靠遗嘱消息 超时兜底那是另一篇的事了。五、面试怎么答这道题如果面试官问你们平台设备消息是怎么处理的我会用一条链路 三个关键词回答设备按约定的 Topicup/产品/设备ID发 JSON 到 EMQX服务端订阅后在 messageArrived 回调里转码并委托给业务层业务层过三层校验设备存在、未禁用、合法 JSON然后把原始 payload 带时间戳落库同时更新设备在线状态。三个关键词契约Topic 先行设备端后端共同遵守、兜底回调吞异常保连接、三层拦截保安全、原始数据先落库再解析为未来的自己留后路。能顺着这条线把为什么讲清楚基本就能让面试官相信消息链路这条主干道你是真的修过。作者人设软件工程在读专升本正在从零搭一个物联网设备接入平台Spring Boot 3.5 EMQX MySQL把踩的每个坑都写成文章。上一篇写了《接口报错全返回 500Spring Boot 全局异常处理与统一响应体实战》欢迎关注追更下一站是消息管道的吞吐升级。
前面的文章把设备怎么连上来一机一密 ACL、断线了怎么办自动重连、遗嘱消息都讲过了。但有一条主线一直没完整写过一条遥测消息从 EMQX 到达服务端之后、落进数据库之前到底经历了什么这条链路是物联网后端的主干道。面试里说说你的平台怎么处理设备消息出现频率极高因为这一个 problem 就能问出你是不是真做过Topic 怎么设计的、消息怎么校验、脏数据怎么办、为什么这么分层。这篇就把我的真实代码摊开讲。一、先约定 Topic它就是设备端的接口契约我平台的消息处理入口是TelemetryMessageHandler.java文件头注释里写着整个系统的主题约定/** * 主题约定 * up/{productKey}/{deviceId} 遥测数据payload 为 JSON * st/{productKey}/{deviceId} 设备状态payload 为 online/offline遗嘱消息用 offline */只有两条主题、三层路径。为什么设计得这么抠因为 Topic 在物联网里就是接口契约——它是设备端和后端共同遵守的 URL定下来就不能随便改。层级越少设备端固件越好写后端解析越不容易出错通配符订阅也越方便服务端只要订阅up/#和st/#两条就能收全所有产品的所有设备。做后端的可以这么类比Topic 是 URL pathpayload 是 request bodyQoS 是要不要重试。把 REST API 设计里契约先行的纪律原样搬过来就行——先定 Topic 和 payload 格式再写解析代码顺序不能反。二、回调函数里第一件事别让一条消息弄死连接消息从 EMQX 推到服务端进入的是 Paho 客户端的回调messageArrived。我在MqttConnection.java里是这么写的OverridepublicvoidmessageArrived(Stringtopic,MqttMessagemessage){StringpayloadnewString(message.getPayload(),StandardCharsets.UTF_8);try{handler.handle(topic,payload);}catch(Exceptione){// 单条消息处理失败不能影响连接和后续消息log.error(处理消息失败: topic{},topic,e);}}重点是这个 try-catch。messageArrived跑在 Paho 客户端单条的回调线程上业务代码在这里抛异常后果不只是丢一条消息——处理流程卡住后续消息全部排队甚至影响心跳和连接稳定性。所以回调里只做两件事转码、委托给业务 handler并且任何异常都必须就地吞掉记日志。连接参数上我也做了取舍这几行在面试里被追问的概率很高options.setAutomaticReconnect(true);options.setCleanSession(true);options.setMaxInflight(100);// 订阅时int[]qos{1,1};QoS 用 1遥测数据允许重复但不允许丢cleanSession 用 true服务端掉线期间的旧消息不强求补齐因为看板要的是当前状态几秒前的旧温湿度补回来反而是脏数据。这是业务属性决定的选型不是拍脑袋。三、三层拦截认证、鉴权、业务校验各拦各的真正进业务逻辑之前每条消息要过三道门。第一道在 EMQX一机一密认证 ACL前面有文章专门写过但后端代码里还有两道DevicedevicedeviceRepository.findById(deviceId).orElse(null);if(devicenull||!productKey.equals(device.getProductKey())){log.warn(收到未知设备的数据已丢弃: topic{},topic);return;}// 已禁用的设备即使 Broker 层有漏网的ACL 未生效/缓存入库前再拦一道。// 安全上的原则认证、鉴权、业务校验三层各拦各的不假设上一层一定生效。if(DeviceService.STATUS_DISABLED.equals(device.getStatus())){log.warn(设备[{}]已禁用丢弃其消息: topic{},deviceId,topic);return;}为什么要重复拦截因为每一层防的风险不一样**第一层认证**防你是谁都没搞清——伪造 clientId、偷来的密钥**第二层ACL**防合法设备发不该发的主题——但 ACL 有缓存、有生效延迟**第三层业务校验**兜住前两层的一切漏网被禁用的设备、topic 里 deviceId 和 productKey 对不上的伪造请求。这段代码注释里那句话是我踩坑之后的总结“不假设上一层一定生效”。分布式系统里每一层都可能有缓存、延迟、配置漂移把安全押在前面那层应该拦住了上迟早出事故。四、先存原始 JSON再谈解析过了三道门的遥测消息我的处理方式可能和很多教程不一样——不做字段级解析原始 payload 直接落库privatevoidhandleTelemetry(Devicedevice,StringproductKey,Stringpayload){// 校验是合法 JSON不合法直接丢弃防脏数据进库try{objectMapper.readTree(payload);}catch(Exceptione){log.warn(payload 不是合法 JSON丢弃: {},payload);return;}device.setStatus(online);deviceRepository.save(device);DataPointdataPointnewDataPoint();dataPoint.setDeviceId(device.getId());dataPoint.setProductKey(productKey);dataPoint.setPayload(payload);// 原样保存不拆字段dataPoint.setReceivedAt(Instant.now());dataPointRepository.save(dataPoint);}两个细节。第一入库前用readTree做一次轻量校验只确认这是合法 JSON不提取任何字段——防的是固件 bug 吐出来的半截数据污染表。第二payload字段存的是原始字符串收到时间单独记一列。为什么这么设计因为设备端是最不可控的一环固件有 bug、传感器会吐异常值、协议后面还会升级。解析逻辑一定会变但原始数据错了就永远错了。只要原始 JSON 在以后想加字段、改解析规则、修历史数据重放一遍就行。做数据平台的人都懂一句话最值钱的不是解析后的结果是带时间戳的原始数据。另外注意device.setStatus(online)这行——收到数据本身就是最可靠的在线信号这比心跳轮询更及时离线判定靠遗嘱消息 超时兜底那是另一篇的事了。五、面试怎么答这道题如果面试官问你们平台设备消息是怎么处理的我会用一条链路 三个关键词回答设备按约定的 Topicup/产品/设备ID发 JSON 到 EMQX服务端订阅后在 messageArrived 回调里转码并委托给业务层业务层过三层校验设备存在、未禁用、合法 JSON然后把原始 payload 带时间戳落库同时更新设备在线状态。三个关键词契约Topic 先行设备端后端共同遵守、兜底回调吞异常保连接、三层拦截保安全、原始数据先落库再解析为未来的自己留后路。能顺着这条线把为什么讲清楚基本就能让面试官相信消息链路这条主干道你是真的修过。作者人设软件工程在读专升本正在从零搭一个物联网设备接入平台Spring Boot 3.5 EMQX MySQL把踩的每个坑都写成文章。上一篇写了《接口报错全返回 500Spring Boot 全局异常处理与统一响应体实战》欢迎关注追更下一站是消息管道的吞吐升级。
