diff --git a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DeviceReportMsgConsumer.java b/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DeviceReportMsgConsumer.java index 0b204348..f95ce0b3 100644 --- a/fastbee-iot-data/src/main/java/com/fastbee/iot/data/consumer/DeviceReportMsgConsumer.java +++ b/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); 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 similarity index 59% rename from fastbee-iot-data/src/main/java/com/fastbee/iot/data/emq/subscribeCallback.java rename to fastbee-iot-data/src/main/java/com/fastbee/iot/data/emq/SubscribeCallback.java index 8618056b..2ede847c 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 @@ -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(); + } } diff --git a/fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/codec/ModbusProtocol.java b/fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/codec/ModbusProtocol.java index 82d799dd..b6977ee0 100644 --- a/fastbee-protocol/fastbee-protocol-collect/src/main/java/com/fastbee/modbus/codec/ModbusProtocol.java +++ b/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();