Administrator
/
FenXiNspJavaGather
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

修复多实例mqttbug

This commit is contained in:
zhangyue 2026-10-22 11:02:34 +08:00
parent 7d6062d539
commit 3068d13cf6
3 changed files with 39 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
src/main/java/com/lanhai/o/iot/lhiot/ResponseCmd.java
View File

@ -9,4 +9,7 @@ import java.util.List;
public class ResponseCmd { public class ResponseCmd {
@ApiModelProperty("指令") @ApiModelProperty("指令")
private ResponseCmdDetail rw_prot; private ResponseCmdDetail rw_prot;
@ApiModelProperty("指令")
private ResponseCmdDetail params;
} }

1
src/main/java/com/lanhai/task/MqttDeviceTask.java
View File

@ -49,6 +49,7 @@ public class MqttDeviceTask {
LambdaQueryWrapper<SurvDeviceDeploy> wrapper = Wrappers.<SurvDeviceDeploy>lambdaQuery(); LambdaQueryWrapper<SurvDeviceDeploy> wrapper = Wrappers.<SurvDeviceDeploy>lambdaQuery();
//监测类设备，并且是mqtt协议 //监测类设备，并且是mqtt协议
wrapper.eq(SurvDeviceDeploy::getRunStatus,0) wrapper.eq(SurvDeviceDeploy::getRunStatus,0)
.ne(SurvDeviceDeploy::getDeviceIotUrl,"")
.eq(SurvDeviceDeploy::getDeployCate,"surv") .eq(SurvDeviceDeploy::getDeployCate,"surv")
.eq(SurvDeviceDeploy::getProtocolType, "mqtt"); .eq(SurvDeviceDeploy::getProtocolType, "mqtt");

39
src/main/java/com/lanhai/util/LhIotUtil.java
View File

@ -28,6 +28,7 @@ public class LhIotUtil {
/** /**
* 设备发送查询指令 * 设备发送查询指令
*/ */
static String clientName = "lhiotMqtt";
public static boolean DeviceQuery(SurvDeviceDeploy survDeviceDeploy){ public static boolean DeviceQuery(SurvDeviceDeploy survDeviceDeploy){
if(survDeviceDeploy!=null){ if(survDeviceDeploy!=null){
@ -35,9 +36,11 @@ public class LhIotUtil {
QueryCmd queryCmd = ConstructCmd(survDeviceDeploy); QueryCmd queryCmd = ConstructCmd(survDeviceDeploy);
//发送指令，传入下行主题 iotUrl //发送指令，传入下行主题 iotUrl
String cmdStr = JSONObject.toJSONString(queryCmd); String cmdStr = JSONObject.toJSONString(queryCmd);
log.error("=================发送查询指令========{}=================",cmdStr);
XxlJobHelper.log("=================发送查询指令========{}=================",cmdStr); XxlJobHelper.log("=========={}=======发送查询指令========{}=================",cmdStr);
mqttService.sendMessage(survDeviceDeploy.getDeviceIotUrl(),cmdStr); boolean b = mqttService.sendMessage(clientName,survDeviceDeploy.getDeviceIotUrl(),cmdStr);
log.error("==========={}======发送查询指令========{}========{}=========",survDeviceDeploy.getDeviceIotUrl(),cmdStr,b);
XxlJobHelper.log("=================发送查询指令re========{}=================",b);
} }
return true; return true;
} }
@ -97,7 +100,35 @@ public class LhIotUtil {
} }
} }
log.warn("{}-----------指令解析结果=============={}",mqttMessage,JSONObject.toJSONString(waterCommonTransVo)); log.warn("{}-----------指令解析结果1=============={}",mqttMessage,JSONObject.toJSONString(waterCommonTransVo));
}else{
log.error("----非主动请求数据，跳过解析==============回执:{}",mqttMessage);
}
}else if(responseCmd.getParams()!=null){
if("up".equals(responseCmd.getParams().getDir())) {
if (zhibiaoList != null && !zhibiaoList.isEmpty()) {
Map<String, String> eleMap = new HashMap<>();
for (ScEquZhibiao scEquZhibiao : zhibiaoList) {
eleMap.put(scEquZhibiao.getCode(), scEquZhibiao.getEntityField());
}
if (responseCmd.getParams().getR_data()!=null && !responseCmd.getParams().getR_data().isEmpty()) {
//因为数据中不包含时间，所以使用当前时间
waterCommonTransVo.setDataDateTime(LocalDateTime.now());
waterCommonTransVo.setDataGatherType(IotConstants.MARK_REALDATA);
for (ResponseCmdData rDatum : responseCmd.getParams().getR_data()) {
String entityName = eleMap.get(rDatum.getName());//字段名称
if (StringUtils.isNotBlank(entityName)) {
TUtil.setFieldValue(waterCommonTransVo, entityName, rDatum.getValue());
counts++;
} else {
log.error("指令2x：{}----未匹配到监测项目:{}", mqttMessage, rDatum.getName());
}
}
}
}
log.warn("{}-----------指令解析结果2=============={}",mqttMessage,JSONObject.toJSONString(waterCommonTransVo));
}else{ }else{
log.error("----非主动请求数据，跳过解析==============回执:{}",mqttMessage); log.error("----非主动请求数据，跳过解析==============回执:{}",mqttMessage);
} }