From ef6693aabbbf12e83d09ad8749f6e60faacc012d Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期二, 29 十一月 2022 15:18:56 +0800
Subject: [PATCH] Merge branch 'wvp-28181-2.0' into Zafu-Dev-1127
---
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisPushStreamStatusMsgListener.java | 51 ++++++++++++++++++++++++++-------------------------
1 files changed, 26 insertions(+), 25 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 4fafa1c..96ff8e8 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
@@ -29,8 +29,6 @@
private final static Logger logger = LoggerFactory.getLogger(RedisPushStreamStatusMsgListener.class);
- private boolean taskQueueHandlerRun = false;
-
@Autowired
private IRedisCatchStorage redisCatchStorage;
@@ -50,37 +48,40 @@
@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鐨凙LARM閫氱煡] 鍙戠幇鏈鐞嗙殑寮傚父, {}",e.getMessage());
}
}
- taskQueueHandlerRun = false;
});
}
}
--
Gitblit v1.8.0