From 54b878d2e1e089cdeb7a957b78aa1f4cbdd6da6c Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期二, 09 四月 2024 22:43:45 +0800
Subject: [PATCH] 优化代码，升级版本号为2.7.1
---
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java | 99 ++++++++++++++++++++++++-------------------------
1 files changed, 48 insertions(+), 51 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
old mode 100644
new mode 100755
index 1897b6f..0912f0b
--- a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisStreamMsgListener.java
@@ -1,11 +1,7 @@
package com.genersoft.iot.vmp.service.redisMsg;
-import com.alibaba.fastjson.JSON;
-import com.alibaba.fastjson.JSONObject;
import com.genersoft.iot.vmp.conf.UserSetting;
-
import com.genersoft.iot.vmp.media.zlm.ZLMMediaListManager;
-import com.genersoft.iot.vmp.media.zlm.dto.MediaItem;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
@@ -33,9 +29,7 @@
@Autowired
private ZLMMediaListManager zlmMediaListManager;
- private boolean taskQueueHandlerRun = false;
-
- private final ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
+ private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
@Qualifier("taskExecutor")
@Autowired
@@ -43,49 +37,52 @@
@Override
public void onMessage(Message message, byte[] bytes) {
-
- taskQueue.offer(message);
- if (!taskQueueHandlerRun) {
- taskQueueHandlerRun = true;
- taskExecutor.execute(() -> {
- while (!taskQueue.isEmpty()) {
- Message msg = taskQueue.poll();
- JSONObject steamMsgJson = JSON.parseObject(msg.getBody(), JSONObject.class);
- if (steamMsgJson == null) {
- logger.warn("[鏀跺埌redis 娴佸彉鍖朷娑堟伅瑙ｆ瀽澶辫触");
- return;
- }
- String serverId = steamMsgJson.getString("serverId");
-
- if (userSetting.getServerId().equals(serverId)) {
- // 鑷繁鍙戦�佺殑娑堟伅蹇界暐鍗冲彲
- return;
- }
- 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");
- MediaItem mediaItem = new MediaItem();
- mediaItem.setSeverId(serverId);
- mediaItem.setApp(app);
- mediaItem.setStream(stream);
- mediaItem.setRegist(register);
- mediaItem.setMediaServerId(mediaServerId);
- mediaItem.setCreateStamp(System.currentTimeMillis()/1000);
- mediaItem.setAliveSecond(0L);
- mediaItem.setTotalReaderCount("0");
- mediaItem.setOriginType(0);
- mediaItem.setOriginTypeStr("0");
- mediaItem.setOriginTypeStr("unknown");
- if (register) {
- zlmMediaListManager.addPush(mediaItem);
- }else {
- zlmMediaListManager.removeMedia(app, stream);
- }
- }
- taskQueueHandlerRun = false;
- });
- }
+// boolean isEmpty = taskQueue.isEmpty();
+// taskQueue.offer(message);
+// if (isEmpty) {
+// taskExecutor.execute(() -> {
+// while (!taskQueue.isEmpty()) {
+// Message msg = taskQueue.poll();
+// 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);
+// }
+// }catch (Exception e) {
+// logger.warn("[REDIS娑堟伅-娴佸彉鍖朷 鍙戠幇鏈鐞嗙殑寮傚父, \r\n{}", JSON.toJSONString(message));
+// logger.error("[REDIS娑堟伅-娴佸彉鍖朷 寮傚父鍐呭锛� ", e);
+// }
+// }
+// });
+// }
}
}
--
Gitblit v1.8.0