Sumuxi/fastbee
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

mqtt直接上报

Browse Source
This commit is contained in:
zhugongping 2026-09-19 11:49:05 +08:00
parent eda5eef6cf
commit bd3335e159
1 changed files with 38 additions and 78 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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();
}
}