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/RedisStreamMsgListener.java | 74 ++++++++++++++++++------------------
1 files changed, 37 insertions(+), 37 deletions(-)
diff --git a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java
index 3e73fc0..1cdc527 100644
--- a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java
@@ -33,8 +33,6 @@
@Autowired
private ZLMMediaListManager zlmMediaListManager;
- private boolean taskQueueHandlerRun = false;
-
private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
@Qualifier("taskExecutor")
@@ -43,48 +41,50 @@
@Override
public void onMessage(Message message, byte[] bytes) {
-
+ boolean isEmpty = taskQueue.isEmpty();
taskQueue.offer(message);
- if (!taskQueueHandlerRun) {
- taskQueueHandlerRun = true;
+ if (isEmpty) {
taskExecutor.execute(() -> {
while (!taskQueue.isEmpty()) {
Message msg = taskQueue.poll();
- JSONObject steamMsgJson = JSON.parseObject(msg.getBody(), JSONObject.class);
- if (steamMsgJson == null) {
- logger.warn("[鏀跺埌redis 娴佸彉鍖朷娑堟伅瑙ｆ瀽澶辫触");
- continue;
- }
- String serverId = steamMsgJson.getString("serverId");
+ try {
+ JSONObject steamMsgJson = JSON.parseObject(msg.getBody(), JSONObject.class);
+ if (steamMsgJson == null) {
+ logger.warn("[鏀跺埌redis 娴佸彉鍖朷娑堟伅瑙ｆ瀽澶辫触");
+ continue;
+ }
+ String serverId = steamMsgJson.getString("serverId");
- if (userSetting.getServerId().equals(serverId)) {
- // 鑷繁鍙戦�佺殑娑堟伅蹇界暐鍗冲彲
- continue;
- }
- logger.info("[鏀跺埌redis 娴佸彉鍖朷锛� {}", new String(message.getBody()));
- String app = steamMsgJson.getString("app");
- String stream = steamMsgJson.getString("stream");
- boolean register = steamMsgJson.getBoolean("register");
- String mediaServerId = steamMsgJson.getString("mediaServerId");
- OnStreamChangedHookParam onStreamChangedHookParam = new OnStreamChangedHookParam();
- onStreamChangedHookParam.setSeverId(serverId);
- onStreamChangedHookParam.setApp(app);
- onStreamChangedHookParam.setStream(stream);
- onStreamChangedHookParam.setRegist(register);
- onStreamChangedHookParam.setMediaServerId(mediaServerId);
- onStreamChangedHookParam.setCreateStamp(System.currentTimeMillis()/1000);
- onStreamChangedHookParam.setAliveSecond(0L);
- onStreamChangedHookParam.setTotalReaderCount("0");
- onStreamChangedHookParam.setOriginType(0);
- onStreamChangedHookParam.setOriginTypeStr("0");
- onStreamChangedHookParam.setOriginTypeStr("unknown");
- if (register) {
- zlmMediaListManager.addPush(onStreamChangedHookParam);
- }else {
- zlmMediaListManager.removeMedia(app, stream);
+ if (userSetting.getServerId().equals(serverId)) {
+ // 鑷繁鍙戦�佺殑娑堟伅蹇界暐鍗冲彲
+ continue;
+ }
+ logger.info("[鏀跺埌redis 娴佸彉鍖朷锛� {}", new String(message.getBody()));
+ String app = steamMsgJson.getString("app");
+ String stream = steamMsgJson.getString("stream");
+ boolean register = steamMsgJson.getBoolean("register");
+ String mediaServerId = steamMsgJson.getString("mediaServerId");
+ OnStreamChangedHookParam onStreamChangedHookParam = new OnStreamChangedHookParam();
+ onStreamChangedHookParam.setSeverId(serverId);
+ onStreamChangedHookParam.setApp(app);
+ onStreamChangedHookParam.setStream(stream);
+ onStreamChangedHookParam.setRegist(register);
+ onStreamChangedHookParam.setMediaServerId(mediaServerId);
+ onStreamChangedHookParam.setCreateStamp(System.currentTimeMillis()/1000);
+ onStreamChangedHookParam.setAliveSecond(0L);
+ onStreamChangedHookParam.setTotalReaderCount("0");
+ onStreamChangedHookParam.setOriginType(0);
+ onStreamChangedHookParam.setOriginTypeStr("0");
+ onStreamChangedHookParam.setOriginTypeStr("unknown");
+ if (register) {
+ zlmMediaListManager.addPush(onStreamChangedHookParam);
+ }else {
+ zlmMediaListManager.removeMedia(app, stream);
+ }
+ }catch (Exception e) {
+ logger.warn("[REDIS鐨凙LARM閫氱煡] 鍙戠幇鏈鐞嗙殑寮傚父, {}",e.getMessage());
}
}
- taskQueueHandlerRun = false;
});
}
}
--
Gitblit v1.8.0