diff --git a/emqx-plugin/pom.xml b/emqx-plugin/pom.xml
index 54ad1bc..4b797e4 100644
--- a/emqx-plugin/pom.xml
+++ b/emqx-plugin/pom.xml
@@ -5,7 +5,7 @@
iot-iita-plugins
cc.iotkit.plugins
- 2.10.18
+ 2.10.30
4.0.0
@@ -83,4 +83,4 @@
-
\ No newline at end of file
+
diff --git a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/service/EmqxPlugin.java b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/service/EmqxPlugin.java
index f71bdd2..6da1f37 100644
--- a/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/service/EmqxPlugin.java
+++ b/emqx-plugin/src/main/java/cc/iotkit/plugins/emqx/service/EmqxPlugin.java
@@ -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> 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 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 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 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 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 attrMap = (Map) points.get(0);
List>arrList = (List>) attrMap.get("points");
Map 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 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 attrMap = (Map) points.get(0);
List>arrList = (List>) attrMap.get("points");
@@ -678,3 +738,4 @@ public class EmqxPlugin implements PluginCloseListener, IPlugin, Runnable {
return null;
}
}
+
diff --git a/emqx-plugin/src/main/resources/application.yml b/emqx-plugin/src/main/resources/application.yml
index ca8141e..95d9200 100644
--- a/emqx-plugin/src/main/resources/application.yml
+++ b/emqx-plugin/src/main/resources/application.yml
@@ -1,5 +1,5 @@
plugin:
- runMode: prod
+ runMode: dev
mainPackage: cc.iotkit.plugin
emqx:
diff --git a/pom.xml b/pom.xml
index b486840..4343be7 100644
--- a/pom.xml
+++ b/pom.xml
@@ -20,7 +20,7 @@
2.7.11
- 2.10.18
+ 2.10.30
cc.iotkit.plugins
iot-iita-plugins
@@ -93,4 +93,4 @@
-
\ No newline at end of file
+