20260905:微信小程序-蓝牙采集网关-数据消费

fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DeviceReportMsgConsumer.java

@@ -24,7 +24,7 @@ public class DeviceReportMsgConsumer {
     @Async(FastBeeConstant.TASK.DEVICE_UP_MESSAGE_TASK)
     public void consume(DeviceReportBo bo) {
         try {
-            //处理数据解析
+            // 处理数据解析
             reportMessageService.parseReportMsg(bo);
         } catch (Exception e) {
             log.error("设备主动上报队列监听异常", e);

fastbee-iot-data/src/main/java/com/fastbee/iot/data/emq/SubscribeCallback.java

@@ -21,7 +21,7 @@ import java.util.Objects;
 
 @Slf4j
 @Component
-public class subscribeCallback implements IMqttMessageListener {
+public class SubscribeCallback implements IMqttMessageListener {
     @Resource
     private RuleProcess ruleProcess;
     @Resource
@@ -31,7 +31,26 @@ public class subscribeCallback implements IMqttMessageListener {
 
     @Override
     public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
-        String message = new String(mqttMessage.getPayload());
+        // 获取原始字节数据（可能是 Modbus 二进制数据或十六进制字符串）
+        byte[] payload = mqttMessage.getPayload();
+
+        // 判断数据格式：如果数据包含非十六进制字符，说明是原始二进制数据
+        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);
@@ -45,6 +64,7 @@ public class subscribeCallback implements IMqttMessageListener {
                 && StringUtils.isNotEmpty(context.getTopic())) {
             topic = context.getTopic();
             message = context.getPayload();
+            data = message.getBytes(StandardCharsets.UTF_8);
         }
         String serialNumber = topicsUtils.parseSerialNumber(topic);
         Long productId;
@@ -53,12 +73,11 @@ public class subscribeCallback implements IMqttMessageListener {
         } else {
             productId = topicsUtils.parseProductId(topic);
         }
-//        Long productId = topicsUtils.parseProductId(topic);
         String name = topicsUtils.parseTopicName(topic);
         DeviceReportBo reportBo = DeviceReportBo.builder()
                 .serialNumber(serialNumber)
                 .productId(productId)
-                .data(message.getBytes(StandardCharsets.UTF_8))
+                .data(data)
                 .platformDate(DateUtils.getNowDate())
                 .topicName(topic)
                 .serverType(ServerType.MQTT)
@@ -66,4 +85,32 @@ public class subscribeCallback implements IMqttMessageListener {
         /*将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;
+            }
+        }
+        return true;
+    }
+
+    /**
+     * 字节数组转十六进制字符串
+     */
+    private String bytesToHex(byte[] bytes) {
+        StringBuilder sb = new StringBuilder();
+        for (byte b : bytes) {
+            sb.append(String.format("%02X", b));
+        }
+        return sb.toString();
+    }
 }

fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/codec/ModbusProtocol.java

@@ -74,7 +74,7 @@ public class ModbusProtocol implements IProtocol {
             DeviceReport report = new DeviceReport();
             report.setProductId(deviceData.getProductId());
             report.setSerialNumber(deviceData.getSerialNumber());
-            //原始报文字符串
+            // 原始报文字符串
             String hexDump = ByteBufUtil.hexDump(deviceData.getBuf());
             ModbusRtu message = messageDecoder.decode(deviceData);
             int slaveId = message.getSlaveId();