leiguohui
/
feijialuo-iot-kitparent
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(plugin): 优化EMQX插件功能并增强设备管理

Browse Source 
- 在application.yml中将运行模式从prod改为dev
- 在application-dev.yml中添加generate_statistics配置项以关闭统计信息
- 在AuthVerticle.java中添加MQTT ACL主题路径转换逻辑
- 在DeviceInfoDataCache.java和DeviceInfoDataImpl.java中增加设备状态查询方法
- 在DeviceInfoExpordVo和DeviceInfoImportVo中新增site字段用于设备位置导出导入
- 在DeviceInstallInfo.java中添加多个图片相关字段如泡沫水测漏图片、设备信息图片、门牌号图片
- 在DeviceInstallInfoDataImpl.java中实现findById方法
- 在DeviceInstallInfoExpordVo和DeviceInstallInfoVo中添加切断阀编号和报警器编号导出功能
- 在DeviceInstallInfoServiceImpl.java中完善安装信息查询逻辑
- 在DeviceManagerServiceImpl.java中支持设备状态查询参数
- 在DeviceQueryBo.java中新增deviceStatus字段用于设备状态筛选
- 在feijialuo-emqx-plugin中优化线程池配置并提升并发处理能力
- 在iot-iita-emqx-plugin中调整设备识别规则适配不同平台需求
master
alsszer 6 months ago
parent febc6291b8
commit 2402bd81df
4 changed files with 202 additions and 141 deletions
Show Stats Download Patch File Download Diff File

4
emqx-plugin/pom.xml
Unescape Escape View File

@ -5,7 +5,7 @@
<parent>
<artifactId>iot-iita-plugins</artifactId>
<groupId>cc.iotkit.plugins</groupId>
<version>2.10.18</version>
<version>2.10.30</version>
</parent>
<modelVersion>4.0.0</modelVersion>
@ -83,4 +83,4 @@
</plugins>
</build>
</project>
</project>

333
emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/service/EmqxPlugin.java
Unescape Escape View File

@ -36,10 +36,7 @@ import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.*;
import static cc.iotkit.plugins.emqx.conf.BCDClockGenerator.formatBCDToHex;
import static cc.iotkit.plugins.emqx.conf.BCDClockGenerator.generateBCDTime;
@ -89,6 +86,18 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {
public static final Map<String, Set<String>> CLIENT_DEVICE_MAP = new HashMap<>();
//定义一个线程池，线程池大小时cpu的2倍
// 创建线程池，大小为CPU核心数的2倍
private static ExecutorService statusThreadPool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
// 创建线程池，大小为CPU核心数的2倍
private static ExecutorService threadPool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
// 创建线程池，大小为CPU核心数的2倍
private static ExecutorService sxsbThreadPool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
// 创建线程池，大小为CPU核心数的2倍
private static ExecutorService onlineThreadPool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2);
@PostConstruct
public void init() {
vertx = Vertx.vertx();
@ -175,6 +184,12 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {
log.error("client connect fail: ", s.cause());
}
}).publishHandler(msg -> {
// 获取当前线程对象
Thread currentThread = Thread.currentThread();
// 获取当前线程的 ID
long threadId = currentThread.getId();
long start = System.currentTimeMillis() ;
log.info("threadId=189{}",threadId+":"+start);
String topic = msg.topicName();
log.info("topic={}",topic);
if (topic.contains("/request/")) {
@ -183,15 +198,15 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {
if (topic != null && topic.startsWith("FIGARO/")){
String s = Arrays.asList(topic.split("/")).get(2);
topic = "FIGARO/post/*/"+s;
// log.info("Client received message on [{}] payload [{}] with QoS [{}]", topic, msg.payload().toJsonObject(), msg.qosLevel());
// log.info("Client received message on [{}] payload [{}] with QoS [{}]", topic, msg.payload().toJsonObject(), msg.qosLevel());
}
// JsonObject payload =msg.payload().toJsonObject();
// JsonObject payload = parseFrame(msg.payload().getBytes());
// JsonObject payload =msg.payload().toJsonObject();
// JsonObject payload = parseFrame(msg.payload().getBytes());
JsonObject payload = parseFrame(processToBytes(msg.payload().getBytes()));
System.out.println("payload原始数据++++++++++++++++++++" + payload);
if(ObjectUtil.isEmpty(payload)){
return;
}
if(ObjectUtil.isEmpty(payload)){
return;
}
try {
//客户端连接断开
if (topic.equals("/sys/client/disconnected")) {
@ -203,7 +218,8 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {
if (device == null) {
return;
}
long start0 = System.currentTimeMillis() ;
log.info("threadId=222{}",threadId+":"+start0);
//有消息上报-设备上线
online("*", device.getDeviceName());
@ -213,7 +229,8 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {
if(StringUtils.isBlank(method)){
method = "thing.event.property.post";
}
long start1 = System.currentTimeMillis() ;
log.info("threadId=229{}",threadId+":"+start1);
if (StringUtils.isBlank(method)) {
return;
}
@ -231,94 +248,121 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {
mapRaw.put("rawHex", rawHex);
mapRaw.put("replyHex", replyHex);
Map<String, Object> result = new HashMap<>(mapRaw);
long start2 = System.currentTimeMillis() ;
log.info("threadId=248{}",threadId+":"+start2);
if(ObjectUtils.isNotEmpty(cmd)&& ObjectUtils.isNotEmpty(data)){
if(cmd.equals("0x11")){
//4.1 状态信息上报（命令码0x11）
DeviceStatusParser.parseStatusReportToMap(data).forEach(
(key, value) -> result.merge(key, value, (v1, v2) -> v2) // 冲突时保留后者
);
// params= payload.getJsonObject(DeviceStatusParser.parseStatusReportToMap(data),defParams);
payload.put("params",result);
device.setDeviceName(result.get("imei").toString());
switch(result.get("deviceType").toString()) {
case "0":
device.setProductKey("CEMpmANABN7Tt6Jh");
//家用报警器
break;
case "1":
device.setProductKey("XmXYxjzihseT76As");
//独立式报警器
break;
case "2":
device.setProductKey("bAASX8tBjYQjBGFP");
if(cmd.equals("0x11")){
//4.1 状态信息上报（命令码0x11）
DeviceStatusParser.parseStatusReportToMap(data).forEach(
(key, value) -> result.merge(key, value, (v1, v2) -> v2) // 冲突时保留后者
);
// params= payload.getJsonObject(DeviceStatusParser.parseStatusReportToMap(data),defParams);
payload.put("params",result);
device.setDeviceName(result.get("imei").toString());
switch(result.get("deviceType").toString()) {
case "0":
device.setProductKey("CEMpmANABN7Tt6Jh");
//家用报警器
break;
case "1":
device.setProductKey("XmXYxjzihseT76As");
//独立式报警器
break;
case "2":
device.setProductKey("bAASX8tBjYQjBGFP");
//工商业控制器
break;
case "3":
//动火离人设备
device.setProductKey("WfpZZFkMxxbGfRca");
case "4":
//联动控制箱设备
System.out.println("联动控制箱设备注册++++++++++++++++++++++++++++++++++++++++++");
device.setProductKey("MQFejp7cyDMH3enG");
break;
}
//注册
IDeviceAction action1 = DeviceRegister.builder()
.model("FIGARO")
.id(UUID.randomUUID().toString())
.deviceName(device.getDeviceName())
.productKey(device.getProductKey())
.params(result)
.time(System.currentTimeMillis())
.build();
thingService.post(pluginInfo.getPluginId(), action1);
}else if(cmd.equals("0x12")){
//4.2 燃气报警器物联网模块数据上报（0x12）
GasAlarmDataParser.parseHexDataToList(data).forEach(
(key, value) -> result.merge(key, value, (v1, v2) -> v2));
payload.put("params",result);
// GasAlarmDataParser.parseHexDataToList(data);
}else if(cmd.equals("0x14")){
//动火离人数据上报与事件上报
FireSafetyDataReceiver.handleHexData(data).forEach(
(key, value) -> result.merge(key, value, (v1, v2) -> v2));
payload.put("params",result);
}else if(cmd.equals("0x15")){
//联动控制箱设备状态及参数数据上报
LinkageControlDataParser.parseHexDataToList(data).forEach(
(key, value) -> result.merge(key, value, (v1, v2) -> v2));
payload.put("params",result);
}
else if(cmd.equals("0x23")){
flag=false;
IoTConfigProtocol.parseReadResponse(data).forEach(
(key, value) -> result.merge(key, value, (v1, v2) -> v2));
//读取配置数据
payload.put("params",result);
//读取配置
DeviceConfig config = DeviceConfig.builder()
.id(UUID.randomUUID().toString())
.deviceName(device.getDeviceName())
.productKey(device.getProductKey())
.time(System.currentTimeMillis())
.config(payload.getJsonObject("params", defParams).getMap())
.build();
thingService.post(pluginInfo.getPluginId(), config);
}
else if(cmd.equals("0x24")){
flag=false;
//下发配置数据的回复
// payload.put("params", FireSafetyDataReceiver.handleHexData(data));
}
break;
case "3":
//动火离人设备
device.setProductKey("WfpZZFkMxxbGfRca");
case "4":
//联动控制箱设备
System.out.println("联动控制箱设备注册++++++++++++++++++++++++++++++++++++++++++");
device.setProductKey("MQFejp7cyDMH3enG");
break;
}
//注册
IDeviceAction action1 = DeviceRegister.builder()
.model("FIGARO")
.id(UUID.randomUUID().toString())
.deviceName(device.getDeviceName())
.productKey(device.getProductKey())
.params(result)
.time(System.currentTimeMillis())
.build();
long start3= System.currentTimeMillis() ;
log.info("threadId=291{}",threadId+":"+start3);
statusThreadPool.submit(new Runnable() {
@Override
public void run() {
thingService.post(pluginInfo.getPluginId(), action1);
}
});
long start4= System.currentTimeMillis() ;
log.info("threadId=299{}",threadId+":"+start4);
}else if(cmd.equals("0x12")){
//4.2 燃气报警器物联网模块数据上报（0x12）
GasAlarmDataParser.parseHexDataToList(data).forEach(
(key, value) -> result.merge(key, value, (v1, v2) -> v2));
payload.put("params",result);
long start5= System.currentTimeMillis() ;
log.info("threadId=306{}",threadId+":"+start5);
// GasAlarmDataParser.parseHexDataToList(data);
}else if(cmd.equals("0x14")){
//动火离人数据上报与事件上报
FireSafetyDataReceiver.handleHexData(data).forEach(
(key, value) -> result.merge(key, value, (v1, v2) -> v2));
payload.put("params",result);
long start6= System.currentTimeMillis() ;
log.info("threadId=314{}",threadId+":"+start6);
}else if(cmd.equals("0x15")){
//联动控制箱设备状态及参数数据上报
LinkageControlDataParser.parseHexDataToList(data).forEach(
(key, value) -> result.merge(key, value, (v1, v2) -> v2));
payload.put("params",result);
long start7= System.currentTimeMillis() ;
log.info("threadId=321{}",threadId+":"+start7);
}
else if(cmd.equals("0x23")){
flag=false;
IoTConfigProtocol.parseReadResponse(data).forEach(
(key, value) -> result.merge(key, value, (v1, v2) -> v2));
//读取配置数据
payload.put("params",result);
//读取配置
DeviceConfig config = DeviceConfig.builder()
.id(UUID.randomUUID().toString())
.deviceName(device.getDeviceName())
.productKey(device.getProductKey())
.time(System.currentTimeMillis())
.config(payload.getJsonObject("params", defParams).getMap())
.build();
long start8= System.currentTimeMillis() ;
log.info("threadId=339{}",threadId+":"+start8);
threadPool.submit(new Runnable() {
@Override
public void run() {
thingService.post(pluginInfo.getPluginId(), config);
}
});
long start9= System.currentTimeMillis() ;
log.info("threadId=347{}",threadId+":"+start9);
}
else if(cmd.equals("0x24")){
flag=false;
//下发配置数据的回复
// Map<String, Object> mapConfig = FireSafetyDataReceiver.handleHexConfigData(data);
// if(!mapConfig.isEmpty()) {
// payload.put("params", mapConfig);
// }
// long start10= System.currentTimeMillis() ;
// log.info("threadId=357{}",threadId+":"+start10);
}
}
// params =
// params =
//解析设备数据
}
@ -328,16 +372,15 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {
if (ObjectUtil.isNull(params) || params.isEmpty()){
return;
}
// System.out.println(payload.getJsonObject("data", defParams));
// System.out.println(payload.getJsonObject("data", defParams));
if (ObjectUtils.isNotEmpty(method) && "thing.lifetime.register".equalsIgnoreCase(method)) {
//子设备注册
String subPk = params.getString("productKey");
String subDn = params.getString("deviceName");
String subModel = params.getString("model");
ActionResult regResult = thingService.post(
pluginInfo.getPluginId(),
fillAction(
long start11= System.currentTimeMillis() ;
log.info("threadId=378{}",threadId+":"+start11);
ActionResult regResult = thingService.post(pluginInfo.getPluginId(),fillAction(
SubDeviceRegister.builder()
.productKey(device.getProductKey())
.deviceName(device.getDeviceName())
@ -352,6 +395,8 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {
.build()
)
);
long start12= System.currentTimeMillis() ;
log.info("threadId=395{}",threadId+":"+start12);
if (regResult.getCode() == 0) {
//注册成功
reply(topic, payload, 0);
@ -379,7 +424,8 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {
.params(params.getMap())
.build();
}
long start13= System.currentTimeMillis() ;
log.info("threadId=424{}",threadId+":"+start13);
try {
if(flag) {
reply1(replyHex,device.getDeviceName(), payload);
@ -395,8 +441,17 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {
action.setProductKey(device.getProductKey());
action.setDeviceName(device.getDeviceName());
action.setTime(System.currentTimeMillis());
thingService.post(pluginInfo.getPluginId(), action);
long start14= System.currentTimeMillis() ;
log.info("threadId=441{}",threadId+":"+start14);
IDeviceAction finalAction = action;
sxsbThreadPool.submit(new Runnable() {
@Override
public void run() {
thingService.post(pluginInfo.getPluginId(), finalAction);
}
});
long start15= System.currentTimeMillis() ;
log.info("threadId=450{}",threadId+":"+start15);
} catch (Exception e) {
log.error("message is illegal.", e);
}
@ -429,17 +484,22 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {
return;
}*/
//上线
thingService.post(
pluginInfo.getPluginId(),
fillAction(DeviceStateChange.builder()
.productKey(pk)
.deviceName(dn)
.state(DeviceState.ONLINE)
.build()
)
);
DEVICE_ONLINE.put(dn, true);
onlineThreadPool.submit(new Runnable() {
@Override
public void run() {
//上线
thingService.post(
pluginInfo.getPluginId(),
fillAction(DeviceStateChange.builder()
.productKey(pk)
.deviceName(dn)
.state(DeviceState.ONLINE)
.build()
)
);
DEVICE_ONLINE.put(dn, true);
}
});
}
public void offline(String clientId) {
@ -502,7 +562,7 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {
// System.out.println("BCD Time: " + formatBCDToHex(generateBCDTime()));
String replyMessage = buildFrame("01", "01",
mid, cmd, "00" + formatBCDToHex(generateBCDTime()));
return replyMessage;
return replyMessage;
}
public static String formatHexByte(String input) {
// 移除0x前缀并保留纯十六进制字符
@ -515,8 +575,8 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {
}
/**
* 回复设备
*/
* 回复设备
*/
private void reply1(String replyMessage,String deviceName, JsonObject payload) {
Map<String, Object> payloadReply = new HashMap<>();
if(ObjectUtil.isNull(replyMessage)) {
@ -543,7 +603,7 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {
System.out.println("cmd" + cmd);
// System.out.println(mid + "re++++++++++++++++++++++++++++++++++++++++++++++++++" + cmd);
// System.out.println("BCD Time: " + formatBCDToHex(generateBCDTime()));
replyMessage = buildFrame("01", "01",
replyMessage = buildFrame("01", "01",
mid, cmd, "00" + formatBCDToHex(generateBCDTime()));
}
payloadReply.put("data", replyMessage);
@ -565,11 +625,11 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {
private void reply(String topic, JsonObject payload, int code) {
Map<String, Object> payloadReply = new HashMap<>();
// System.out.println("BCD Time: " + formatBCDToHex(generateBCDTime()));
String replyMessage = buildFrame("01","01",
// System.out.println("BCD Time: " + formatBCDToHex(generateBCDTime()));
String replyMessage = buildFrame("01","01",
"00","01","01"+formatBCDToHex(generateBCDTime()));
if( StringUtils.isBlank(payload.getString("method"))){
if( StringUtils.isBlank(payload.getString("method"))){
payloadReply.put("id", payload.getString("messageID"));
payloadReply.put("method", "thing.event.property.post_reply");
payloadReply.put("code", code);
@ -578,25 +638,25 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {
JsonObject params = null;
JsonObject defParams = JsonObject.mapFrom(new HashMap<>(0));
params = payload.getJsonObject("params", defParams);
// JsonArray attribute = payload.getJsonArray("params", new JsonArray()); // 提供默认值:ml-citation{ref="1" data="citationList"}
// JsonArray attribute = payload.getJsonArray("params", new JsonArray()); // 提供默认值:ml-citation{ref="1" data="citationList"}
JsonArray attribute = new JsonArray();
// JsonObject jsonObject = payload.getMap().get("params").getJsonObject("params", defParams);
attribute.add(payload.getMap().get("params")); // 回退处理为单元素数组:ml-citation{ref="8" data="citationList"}
JsonArray attribute = new JsonArray();
// JsonObject jsonObject = payload.getMap().get("params").getJsonObject("params", defParams);
attribute.add(payload.getMap().get("params")); // 回退处理为单元素数组:ml-citation{ref="8" data="citationList"}
// JsonArray attribute = payload.getJsonArray("params");
// JsonArray attribute = payload.getJsonArray("params");
List points = attribute.getList();
Map<String, Object> attrMap = (Map<String, Object>) points.get(0);
List<Map<String, Object>>arrList = (List<Map<String, Object>>) attrMap.get("points");
Map<String,Object> map = new HashMap<>();
if(ObjectUtils.isEmpty(arrList)){
payloadReply.put("data", attrMap);
}else{
arrList.forEach(s->{
map.put((String) s.get("name"),s.get("value"));
});
payloadReply.put("data", map);
}
if(ObjectUtils.isEmpty(arrList)){
payloadReply.put("data", attrMap);
}else{
arrList.forEach(s->{
map.put((String) s.get("name"),s.get("value"));
});
payloadReply.put("data", map);
}
topic = topic.replace("/s/", "/c/") + "_reply";
}else{
payloadReply.put("id", payload.getString("id"));
@ -623,7 +683,7 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {
public Map<String, Object> getConfigMap(JsonObject payload) {
JsonArray attribute = new JsonArray();
attribute.add(payload.getMap().get("params"));
// JsonArray attribute = payload.getJsonArray("params");
// JsonArray attribute = payload.getJsonArray("params");
List points = attribute.getList();
Map<String, Object> attrMap = (Map<String, Object>) points.get(0);
List<Map<String, Object>>arrList = (List<Map<String, Object>>) attrMap.get("points");
@ -678,3 +738,4 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {
return null;
}
}

2
emqx-plugin/src/main/resources/application.yml
Unescape Escape View File

@ -1,5 +1,5 @@
plugin:
runMode: prod
runMode: dev
mainPackage: cc.iotkit.plugin
emqx:

4
pom.xml
Unescape Escape View File

@ -20,7 +20,7 @@
<version>2.7.11</version>
<relativePath/>
</parent>
<version>2.10.18</version>
<version>2.10.30</version>
<groupId>cc.iotkit.plugins</groupId>
<artifactId>iot-iita-plugins</artifactId>
@ -93,4 +93,4 @@
</dependencies>
</project>
</project>

Write Preview
Loading…
Cancel
Save