From 534be3f5809f430cb46cb0fcbba99d3d425f2324 Mon Sep 17 00:00:00 2001
From: 648540858 <648540858@qq.com>
Date: 星期三, 07 六月 2023 20:58:35 +0800
Subject: [PATCH] 支持redis消息强制关闭流
---
src/main/java/com/genersoft/iot/vmp/conf/redis/RedisMsgListenConfig.java | 4 ++
src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java | 1
src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisCloseStreamMsgListener.java | 59 +++++++++++++++++++++++++++++
src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java | 5 ++
4 files changed, 69 insertions(+), 0 deletions(-)
diff --git a/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java b/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java
index 51c2ab1..a06c6d0 100644
--- a/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java
+++ b/src/main/java/com/genersoft/iot/vmp/common/VideoManagerConstants.java
@@ -107,6 +107,11 @@
public static final String VM_MSG_STREAM_PUSH_RESPONSE = "VM_MSG_STREAM_PUSH_RESPONSE";
/**
+ * redis 閫氱煡骞冲彴鍏抽棴鎺ㄦ祦
+ */
+ public static final String VM_MSG_STREAM_PUSH_CLOSE = "VM_MSG_STREAM_PUSH_CLOSE";
+
+ /**
* redis 娑堟伅璇锋眰鎵�鏈夌殑鍦ㄧ嚎閫氶亾
*/
public static final String VM_MSG_GET_ALL_ONLINE_REQUESTED = "VM_MSG_GET_ALL_ONLINE_REQUESTED";
diff --git a/src/main/java/com/genersoft/iot/vmp/conf/redis/RedisMsgListenConfig.java b/src/main/java/com/genersoft/iot/vmp/conf/redis/RedisMsgListenConfig.java
index 9f48426..7e1cc1d 100644
--- a/src/main/java/com/genersoft/iot/vmp/conf/redis/RedisMsgListenConfig.java
+++ b/src/main/java/com/genersoft/iot/vmp/conf/redis/RedisMsgListenConfig.java
@@ -43,6 +43,9 @@
@Autowired
private RedisPushStreamResponseListener redisPushStreamResponseListener;
+ @Autowired
+ private RedisCloseStreamMsgListener redisCloseStreamMsgListener;
+
/**
* redis娑堟伅鐩戝惉鍣ㄥ鍣� 鍙互娣诲姞澶氫釜鐩戝惉涓嶅悓璇濋鐨剅edis鐩戝惉鍣紝鍙渶瑕佹妸娑堟伅鐩戝惉鍣ㄥ拰鐩稿簲鐨勬秷鎭闃呭鐞嗗櫒缁戝畾锛岃娑堟伅鐩戝惉鍣�
@@ -63,6 +66,7 @@
container.addMessageListener(redisPushStreamStatusMsgListener, new PatternTopic(VideoManagerConstants.VM_MSG_PUSH_STREAM_STATUS_CHANGE));
container.addMessageListener(redisPushStreamListMsgListener, new PatternTopic(VideoManagerConstants.VM_MSG_PUSH_STREAM_LIST_CHANGE));
container.addMessageListener(redisPushStreamResponseListener, new PatternTopic(VideoManagerConstants.VM_MSG_STREAM_PUSH_RESPONSE));
+ container.addMessageListener(redisCloseStreamMsgListener, new PatternTopic(VideoManagerConstants.VM_MSG_STREAM_PUSH_CLOSE));
return container;
}
}
diff --git a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java
index 0a03c66..dcaab9e 100644
--- a/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java
+++ b/src/main/java/com/genersoft/iot/vmp/service/impl/StreamPushServiceImpl.java
@@ -183,6 +183,7 @@
@Override
public boolean stop(String app, String streamId) {
+ logger.info("[鎺ㄦ祦 ] 鍋滄娴侊細 {}/{}", app, streamId);
StreamPushItem streamPushItem = streamPushMapper.selectOne(app, streamId);
if (streamPushItem != null) {
gbStreamService.sendCatalogMsg(streamPushItem, CatalogEvent.DEL);
diff --git a/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisCloseStreamMsgListener.java b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisCloseStreamMsgListener.java
new file mode 100644
index 0000000..d010475
--- /dev/null
+++ b/src/main/java/com/genersoft/iot/vmp/service/redisMsg/RedisCloseStreamMsgListener.java
@@ -0,0 +1,59 @@
+package com.genersoft.iot.vmp.service.redisMsg;
+
+import com.alibaba.fastjson2.JSON;
+import com.alibaba.fastjson2.JSONObject;
+import com.genersoft.iot.vmp.service.IStreamPushService;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.connection.MessageListener;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.ConcurrentLinkedQueue;
+
+/**
+ * 鎺ユ敹鏉ヨ嚜redis鐨勫叧闂祦鏇存柊閫氱煡
+ * @author lin
+ */
+@Component
+public class RedisCloseStreamMsgListener implements MessageListener {
+
+ private final static Logger logger = LoggerFactory.getLogger(RedisCloseStreamMsgListener.class);
+
+
+ @Autowired
+ private IStreamPushService pushService;
+
+ private ConcurrentLinkedQueue<Message> taskQueue = new ConcurrentLinkedQueue<>();
+
+ @Qualifier("taskExecutor")
+ @Autowired
+ private ThreadPoolTaskExecutor taskExecutor;
+
+ @Override
+ public void onMessage(@NotNull Message message, byte[] bytes) {
+ boolean isEmpty = taskQueue.isEmpty();
+ taskQueue.offer(message);
+ if (isEmpty) {
+ taskExecutor.execute(() -> {
+ while (!taskQueue.isEmpty()) {
+ Message msg = taskQueue.poll();
+ try {
+ JSONObject jsonObject = JSON.parseObject(msg.getBody());
+ String app = jsonObject.getString("app");
+ String stream = jsonObject.getString("stream");
+ pushService.stop(app, stream);
+
+ }catch (Exception e) {
+ logger.warn("[REDIS鐨勫叧闂帹娴侀�氱煡] 鍙戠幇鏈鐞嗙殑寮傚父, \r\n{}", JSON.toJSONString(message));
+ logger.error("[REDIS鐨勫叧闂帹娴侀�氱煡] 寮傚父鍐呭锛� ", e);
+ }
+ }
+ });
+ }
+ }
+}
--
Gitblit v1.8.0