From c3ce2bc5d069ba57309567aee4ae418bc0cf75ed Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期五, 17 三月 2023 13:22:05 +0800
Subject: [PATCH] 优化异常处理以及级联设备状态查询
---
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamStatusMsgListener.java | 54 ++++++++++++++++++++++++++++--------------------------
1 files changed, 28 insertions(+), 26 deletions(-)
diff --git a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamStatusMsgListener.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamStatusMsgListener.java
index 2faf3b8..d7e02f5 100644
--- a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamStatusMsgListener.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamStatusMsgListener.java
@@ -1,6 +1,6 @@
package com.genersoft.iot.vmp.service.redisMsg;
-import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson2.JSON;
import com.genersoft.iot.vmp.common.VideoManagerConstants;
import com.genersoft.iot.vmp.conf.DynamicTask;
import com.genersoft.iot.vmp.service.IStreamPushService;
@@ -29,8 +29,6 @@
private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamStatusMsgListener.class);
- private boolean taskQueueHandlerRun = false;
-
@Autowired
private IRedisCatchStorage redisCatchStorage;
@@ -50,37 +48,41 @@
@Override
public void onMessage(Message message, byte[] bytes) {
+ boolean isEmpty = taskQueue.isEmpty();
logger.warn("[REDIS娑堟伅-鎺ㄦ祦璁惧鐘舵�佸彉鍖朷锛� {}", new String(message.getBody()));
taskQueue.offer(message);
- if (!taskQueueHandlerRun) {
- taskQueueHandlerRun = true;
+ if (isEmpty) {
taskExecutor.execute(() -> {
while (!taskQueue.isEmpty()) {
Message msg = taskQueue.poll();
- PushStreamStatusChangeFromRedisDto statusChangeFromPushStream = JSON.parseObject(msg.getBody(), PushStreamStatusChangeFromRedisDto.class);
- if (statusChangeFromPushStream == null) {
- logger.warn("[REDIS娑堟伅]鎺ㄦ祦璁惧鐘舵�佸彉鍖栨秷鎭В鏋愬け璐�");
- continue;
- }
- // 鍙栨秷瀹氭椂浠诲姟
- dynamicTask.stop(VideoManagerConstants.VM_MSG_GET_ALL_ONLINE_REQUESTED);
- if (statusChangeFromPushStream.isSetAllOffline()) {
- // 鎵�鏈夎澶囩绾�
- streamPushService.allStreamOffline();
- }
- if (statusChangeFromPushStream.getOfflineStreams() != null
- && statusChangeFromPushStream.getOfflineStreams().size() > 0) {
- // 鏇存柊閮ㄥ垎璁惧绂荤嚎
- streamPushService.offline(statusChangeFromPushStream.getOfflineStreams());
- }
- if (statusChangeFromPushStream.getOnlineStreams() != null &&
- statusChangeFromPushStream.getOnlineStreams().size() > 0) {
- // 鏇存柊閮ㄥ垎璁惧涓婄嚎
- streamPushService.online(statusChangeFromPushStream.getOnlineStreams());
+ try {
+ PushStreamStatusChangeFromRedisDto statusChangeFromPushStream = JSON.parseObject(msg.getBody(), PushStreamStatusChangeFromRedisDto.class);
+ if (statusChangeFromPushStream == null) {
+ logger.warn("[REDIS娑堟伅]鎺ㄦ祦璁惧鐘舵�佸彉鍖栨秷鎭В鏋愬け璐�");
+ continue;
+ }
+ // 鍙栨秷瀹氭椂浠诲姟
+ dynamicTask.stop(VideoManagerConstants.VM_MSG_GET_ALL_ONLINE_REQUESTED);
+ if (statusChangeFromPushStream.isSetAllOffline()) {
+ // 鎵�鏈夎澶囩绾�
+ streamPushService.allStreamOffline();
+ }
+ if (statusChangeFromPushStream.getOfflineStreams() != null
+ && statusChangeFromPushStream.getOfflineStreams().size() > 0) {
+ // 鏇存柊閮ㄥ垎璁惧绂荤嚎
+ streamPushService.offline(statusChangeFromPushStream.getOfflineStreams());
+ }
+ if (statusChangeFromPushStream.getOnlineStreams() != null &&
+ statusChangeFromPushStream.getOnlineStreams().size() > 0) {
+ // 鏇存柊閮ㄥ垎璁惧涓婄嚎
+ streamPushService.online(statusChangeFromPushStream.getOnlineStreams());
+ }
+ }catch (Exception e) {
+ logger.warn("[REDIS娑堟伅-鎺ㄦ祦璁惧鐘舵�佸彉鍖朷 鍙戠幇鏈鐞嗙殑寮傚父, \r\n{}", JSON.toJSONString(message));
+ logger.error("[REDIS娑堟伅-鎺ㄦ祦璁惧鐘舵�佸彉鍖朷 寮傚父鍐呭锛� ", e);
}
}
- taskQueueHandlerRun = false;
});
}
}
--
Gitblit v1.8.0