diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/emq/SubscribeCallback.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/emq/SubscribeCallback.java index 2ede847c..93305e42 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/emq/SubscribeCallback.java +++ b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/emq/SubscribeCallback.java @@ -3,6 +3,7 @@ package com.fastbee.iot.data.emq; import com.fastbee.common.core.mq.DeviceReportBo; import com.fastbee.common.enums.ServerType; import com.fastbee.common.enums.TopicType; +import com.fastbee.common.trace.TraceContext; import com.fastbee.common.utils.DateUtils; import com.fastbee.common.utils.StringUtils; import com.fastbee.common.utils.gateway.mq.TopicsUtils; @@ -31,86 +32,45 @@ public class SubscribeCallback implements IMqttMessageListener { @Override public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception { - // 获取原始字节数据(可能是 Modbus 二进制数据或十六进制字符串) - byte[] payload = mqttMessage.getPayload(); + try { + // 在消息入口最前端生成traceId,保证"接收消息主题/Qos/内容"等最前面的日志也携带traceId + TraceContext.getOrCreateTraceId(); + String message = new String(mqttMessage.getPayload()); + log.info("接收消息主题 : {}", topic); + log.info("接收消息Qos : {}", mqttMessage.getQos()); + log.info("接收消息内容 : {}", message); - // 判断数据格式:如果数据包含非十六进制字符,说明是原始二进制数据 - boolean isHexString = isHexString(payload); - - byte[] data; - String message; - - if (isHexString) { - // 十六进制字符串格式(Modbus 小程序蓝牙采集) - message = new String(payload, StandardCharsets.UTF_8); - data = message.getBytes(StandardCharsets.UTF_8); - log.info("接收消息格式: 十六进制字符串"); - } else { - // 二进制格式(兼容旧版本) - message = bytesToHex(payload); - data = payload; - log.info("接收消息格式: 二进制数据"); - } - log.info("接收消息主题 : {}", topic); - log.info("接收消息Qos : {}", mqttMessage.getQos()); - log.info("接收消息内容 : {}", message); - - //这里默认设备编号长度超过9位 - String[] split = topic.split("/"); - String sn = Arrays.stream(split).filter(imei -> imei.length() > 5).findFirst().get(); - // 规则引擎脚本处理,完成后返回结果 - MsgContext context = ruleProcess.processRuleScript(sn, 1, topic, message); - if (!Objects.isNull(context) && StringUtils.isNotEmpty(context.getPayload()) - && StringUtils.isNotEmpty(context.getTopic())) { - topic = context.getTopic(); - message = context.getPayload(); - data = message.getBytes(StandardCharsets.UTF_8); - } - String serialNumber = topicsUtils.parseSerialNumber(topic); - Long productId; - if (topic.contains(TopicType.FETCH_UPGRADE_REPLY.getTopicSuffix()) || topic.contains(TopicType.HTTP_UPGRADE_REPLY.getTopicSuffix())) { - productId = 0L; - } else { - productId = topicsUtils.parseProductId(topic); - } - String name = topicsUtils.parseTopicName(topic); - DeviceReportBo reportBo = DeviceReportBo.builder() - .serialNumber(serialNumber) - .productId(productId) - .data(data) - .platformDate(DateUtils.getNowDate()) - .topicName(topic) - .serverType(ServerType.MQTT) - .build(); - /*将mqtt的消息发送至MQ队列处理消息 ,减轻mqtt客户端消息压力*/ - emqxMessageSerice.sendEmqxMessage(name, reportBo); - } - - /** - * 判断字节数组是否为十六进制字符串 - * 十六进制字符串只包含 0-9, A-F, a-f 字符 - */ - private boolean isHexString(byte[] bytes) { - if (bytes == null || bytes.length == 0) { - return false; - } - for (byte b : bytes) { - // 判断是否为十六进制字符(0-9, A-F, a-f) - if (!((b >= '0' && b <= '9') || (b >= 'A' && b <= 'F') || (b >= 'a' && b <= 'f'))) { - return false; + //这里默认设备编号长度超过9位 + String[] split = topic.split("/"); + String sn = Arrays.stream(split).filter(imei -> imei.length() > 5).findFirst().get(); + // 规则引擎脚本处理,完成后返回结果 + MsgContext context = ruleProcess.processRuleScript(sn, 1, topic, message); + if (!Objects.isNull(context) && StringUtils.isNotEmpty(context.getPayload()) + && StringUtils.isNotEmpty(context.getTopic())) { + topic = context.getTopic(); + message = context.getPayload(); } + String serialNumber = topicsUtils.parseSerialNumber(topic); + Long productId; + if (topic.contains(TopicType.FETCH_UPGRADE_REPLY.getTopicSuffix()) || topic.contains(TopicType.HTTP_UPGRADE_REPLY.getTopicSuffix())) { + productId = 0L; + } else { + productId = topicsUtils.parseProductId(topic); + } + String name = topicsUtils.parseTopicName(topic); + DeviceReportBo reportBo = DeviceReportBo.builder() + .serialNumber(serialNumber) + .productId(productId) + .data(message.getBytes(StandardCharsets.UTF_8)) + .platformDate(DateUtils.getNowDate()) + .topicName(topic) + .serverType(ServerType.MQTT) + .build(); + /*将mqtt的消息发送至MQ队列处理消息 ,减轻mqtt客户端消息压力*/ + emqxMessageSerice.sendEmqxMessage(name, reportBo); + } finally { + // MQTT客户端回调线程是线程池复用的,必须清理,防止traceId串到无关消息上 + TraceContext.clear(); } - return true; - } - - /** - * 字节数组转十六进制字符串 - */ - private String bytesToHex(byte[] bytes) { - StringBuilder sb = new StringBuilder(); - for (byte b : bytes) { - sb.append(String.format("%02X", b)); - } - return sb.toString(); } }