代码优化

zc-business/src/main/java/com/zc/business/message/device/handler/DeviceMessageHandler.java

@@ -0,0 +1,37 @@
+package com.zc.business.message.device.handler;
+
+import com.zc.business.domain.DcDevice;
+import com.zc.business.service.IDcDeviceService;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * 设备消息处理
+ */
+@Service
+public class DeviceMessageHandler {
+
+    @Resource
+    private IDcDeviceService dcDeviceService;
+
+    /**
+     * 更新设备状态
+     * @param msg 物联设备id集合
+     * @param state 设备状态 0 异常 1 正常
+     */
+    public void updateDeviceState(List<String> msg, String state) {
+
+        List<DcDevice> dcDevices =  msg.stream().map(iotDeviceId -> {
+            DcDevice dcDevice = new DcDevice();
+            dcDevice.setDeviceState(state);
+            dcDevice.setIotDeviceId(iotDeviceId);
+            return dcDevice;
+        }).collect(Collectors.toList());
+
+        // 批量更新设备状态
+        dcDeviceService.batchUpdate(dcDevices);
+    }
+}

zc-business/src/main/java/com/zc/business/message/device/DeviceEventListener.java → zc-business/src/main/java/com/zc/business/message/device/listener/DeviceEventListener.java

@@ -1,5 +1,7 @@
-package com.zc.business.message.device;
+package com.zc.business.message.device.listener;
 
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
 import com.zc.common.core.redis.stream.RedisStream;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -7,8 +9,11 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.connection.stream.ObjectRecord;
 import org.springframework.data.redis.connection.stream.RecordId;
 import org.springframework.data.redis.stream.StreamListener;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.stereotype.Component;
 
+import javax.annotation.Resource;
+
 /**
  * 设备消息监听
  */
@@ -20,12 +25,24 @@ public class DeviceEventListener implements StreamListener<String, ObjectRecord<
     @Autowired
     private RedisStream redisStream;
 
+    @Resource
+    private ThreadPoolTaskExecutor threadPoolTaskExecutor;
+
     @Override
     public void onMessage(ObjectRecord<String, String> message) {
         String streamKay = message.getStream();
         RecordId recordId = message.getId();
 
+        threadPoolTaskExecutor.execute(() -> {
+            JSONObject data = JSON.parseObject(message.getValue(), JSONObject.class);
+            this.handle(data);
+        });
+
         // 消费完后直接删除消息
         redisStream.del(streamKay, String.valueOf(recordId));
     }
+
+    private void handle(JSONObject data) {
+        System.out.println(data);
+    }
 }

zc-business/src/main/java/com/zc/business/message/device/DeviceFunctionReplyListener.java → zc-business/src/main/java/com/zc/business/message/device/listener/DeviceFunctionReplyListener.java

@@ -1,4 +1,4 @@
-package com.zc.business.message.device;
+package com.zc.business.message.device.listener;
 
 import com.zc.common.core.redis.stream.RedisStream;
 import org.slf4j.Logger;

zc-business/src/main/java/com/zc/business/message/device/DevicePropertyReadReplyListener.java → zc-business/src/main/java/com/zc/business/message/device/listener/DevicePropertyReadReplyListener.java

@@ -1,4 +1,4 @@
-package com.zc.business.message.device;
+package com.zc.business.message.device.listener;
 
 import com.zc.common.core.redis.stream.RedisStream;
 import org.slf4j.Logger;

zc-business/src/main/java/com/zc/business/message/device/DevicePropertyReportListener.java → zc-business/src/main/java/com/zc/business/message/device/listener/DevicePropertyReportListener.java

@@ -1,4 +1,4 @@
-package com.zc.business.message.device;
+package com.zc.business.message.device.listener;
 
 import com.zc.common.core.redis.stream.RedisStream;
 import org.slf4j.Logger;

zc-business/src/main/java/com/zc/business/message/device/DevicePropertyWriteReplyListener.java → zc-business/src/main/java/com/zc/business/message/device/listener/DevicePropertyWriteReplyListener.java

@@ -1,4 +1,4 @@
-package com.zc.business.message.device;
+package com.zc.business.message.device.listener;
 
 import com.zc.common.core.redis.stream.RedisStream;
 import org.slf4j.Logger;

zc-business/src/main/java/com/zc/business/message/device/OfflineMessageListener.java → zc-business/src/main/java/com/zc/business/message/device/listener/OfflineMessageListener.java

@@ -1,7 +1,8 @@
-package com.zc.business.message.device;
+package com.zc.business.message.device.listener;
 
 import com.alibaba.fastjson.JSON;
 import com.zc.business.domain.DcDevice;
+import com.zc.business.message.device.handler.DeviceMessageHandler;
 import com.zc.business.service.IDcDeviceService;
 import com.zc.common.core.redis.stream.RedisStream;
 import org.slf4j.Logger;
@@ -28,10 +29,10 @@ public class OfflineMessageListener implements StreamListener<String, ObjectReco
     private RedisStream redisStream;
 
     @Resource
-    private IDcDeviceService dcDeviceService;
+    private ThreadPoolTaskExecutor threadPoolTaskExecutor;
 
     @Resource
-    private ThreadPoolTaskExecutor threadPoolTaskExecutor;
+    private DeviceMessageHandler deviceMessageHandler;
 
     @Override
     public void onMessage(ObjectRecord<String, String> message) {
@@ -39,28 +40,12 @@ public class OfflineMessageListener implements StreamListener<String, ObjectReco
         RecordId recordId = message.getId();
 
         threadPoolTaskExecutor.execute(() -> {
-            List<String> list = JSON.parseArray(message.getValue(), String.class);
-            this.handle(list);
+            List<String> data = JSON.parseArray(message.getValue(), String.class);
+            deviceMessageHandler.updateDeviceState(data, DcDevice.OFFLINE);
         });
 
         // 消费完后直接删除消息
         redisStream.del(streamKay, String.valueOf(recordId));
     }
 
-    /**
-     * 处理离线消息
-     * @param msg 离线消息
-     */
-    private void handle(List<String> msg) {
-
-        List<DcDevice> dcDevices =  msg.stream().map(iotDeviceId -> {
-            DcDevice dcDevice = new DcDevice();
-            dcDevice.setDeviceState(DcDevice.OFFLINE);
-            dcDevice.setIotDeviceId(iotDeviceId);
-            return dcDevice;
-        }).collect(Collectors.toList());
-
-        // 批量更新设备状态
-        dcDeviceService.batchUpdate(dcDevices);
-    }
 }

zc-business/src/main/java/com/zc/business/message/device/OnlineMessageListener.java → zc-business/src/main/java/com/zc/business/message/device/listener/OnlineMessageListener.java

@@ -1,7 +1,8 @@
-package com.zc.business.message.device;
+package com.zc.business.message.device.listener;
 
 import com.alibaba.fastjson.JSON;
 import com.zc.business.domain.DcDevice;
+import com.zc.business.message.device.handler.DeviceMessageHandler;
 import com.zc.business.service.IDcDeviceService;
 import com.zc.common.core.redis.stream.RedisStream;
 import org.slf4j.Logger;
@@ -15,7 +16,6 @@ import org.springframework.stereotype.Component;
 
 import javax.annotation.Resource;
 import java.util.List;
-import java.util.stream.Collectors;
 
 /**
  * 在线消息监听
@@ -29,7 +29,7 @@ public class OnlineMessageListener implements StreamListener<String, ObjectRecor
     private RedisStream redisStream;
 
     @Resource
-    private IDcDeviceService dcDeviceService;
+    private DeviceMessageHandler deviceMessageHandler;
 
     @Resource
     private ThreadPoolTaskExecutor threadPoolTaskExecutor;
@@ -40,28 +40,11 @@ public class OnlineMessageListener implements StreamListener<String, ObjectRecor
         RecordId recordId = message.getId();
 
         threadPoolTaskExecutor.execute(() -> {
-            List<String> list = JSON.parseArray(message.getValue(), String.class);
-            this.handle(list);
+            List<String> data = JSON.parseArray(message.getValue(), String.class);
+            deviceMessageHandler.updateDeviceState(data, DcDevice.ONLINE);
         });
 
         // 消费完后直接删除消息
         redisStream.del(streamKay, String.valueOf(recordId));
     }
-
-
-    /**
-     * 处理在线消息
-     * @param msg 在线消息
-     */
-    private void handle(List<String> msg) {
-        List<DcDevice> dcDevices =  msg.stream().map(iotDeviceId -> {
-            DcDevice dcDevice = new DcDevice();
-            dcDevice.setDeviceState(DcDevice.ONLINE);
-            dcDevice.setIotDeviceId(iotDeviceId);
-            return dcDevice;
-        }).collect(Collectors.toList());
-
-        // 批量更新设备状态
-        dcDeviceService.batchUpdate(dcDevices);
-    }
 }

zc-business/src/main/java/com/zc/business/message/device/MessageSubscription.java → zc-business/src/main/java/com/zc/business/message/device/subscribe/MessageSubscription.java

@@ -1,6 +1,7 @@
-package com.zc.business.message.device;
+package com.zc.business.message.device.subscribe;
 
 import com.zc.business.constant.RedisStreamConstants;
+import com.zc.business.message.device.listener.*;
 import com.zc.common.core.redis.stream.RedisStream;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;