From a2bef1bc001cc196affb1b24d69eec4127cfa40a Mon Sep 17 00:00:00 2001
From: liusuyi <1951119284@qq.com>
Date: Wed, 01 Apr 2026 10:04:55 +0800
Subject: [PATCH] 优化
---
ard-vtdu/src/main/java/com/ard/vtdu/service/impl/ArdVtduServiceImpl.java | 166 ++++++++++++++++++++++++++++++++++++++++++-------------
1 files changed, 126 insertions(+), 40 deletions(-)
diff --git a/ard-vtdu/src/main/java/com/ard/vtdu/service/impl/ArdVtduServiceImpl.java b/ard-vtdu/src/main/java/com/ard/vtdu/service/impl/ArdVtduServiceImpl.java
index 0aebb02..62f3855 100644
--- a/ard-vtdu/src/main/java/com/ard/vtdu/service/impl/ArdVtduServiceImpl.java
+++ b/ard-vtdu/src/main/java/com/ard/vtdu/service/impl/ArdVtduServiceImpl.java
@@ -1,5 +1,7 @@
package com.ard.vtdu.service.impl;
+import com.ard.common.constant.CacheConstants;
+import com.ard.common.core.redis.RedisCache;
import com.ard.common.utils.DateUtils;
import com.ard.vtdu.domain.ArdVtdu;
import com.ard.vtdu.domain.Conf;
@@ -7,13 +9,17 @@
import com.ard.vtdu.service.IArdVtduService;
import com.ard.vtdu.service.IArdMediaService;
import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.security.authentication.event.AuthenticationSuccessEvent;
+import com.ard.common.core.domain.model.LoginUser;
+import org.springframework.security.authentication.event.LogoutSuccessEvent;
+import org.springframework.security.core.Authentication;
import org.springframework.stereotype.Service;
+
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.*;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
import java.util.stream.Collectors;
/**
@@ -28,22 +34,14 @@
@Resource
private ArdVtduMapper ardVtduMapper;
@Resource
- private IArdMediaService mediaService;
-
-
+ private IArdMediaService ardMediaService;
+ @Resource
+ private RedisCache redisCache;
@PostConstruct
public void initMediaMTX() {
//清空mediamtx流媒体数据
- mediaService.clearAll();
- }
-
- @Scheduled(cron = "0/5 * * * * ?")
- public void syncMediamtx() {
- log.debug("定时同步MediaMTX任务");
- List<String> nameList = mediaService.getNameList();
- List<ArdVtdu> ardVtduList = selectArdVtduList(new ArdVtdu());
- asyncMediaMTX(ardVtduList, nameList);
+ ardMediaService.clearAll();
}
/**
@@ -76,7 +74,8 @@
*/
@Override
public int insertArdVtdu(ArdVtdu ardVtdu) {
- ardVtdu = mediaService.addPath(ardVtdu);
+ ardVtdu.setCreateTime(DateUtils.getNowDate());
+ ardVtdu = ardMediaService.addPath(ardVtdu);
return ardVtduMapper.insertArdVtdu(ardVtdu);
}
@@ -88,7 +87,8 @@
*/
@Override
public int updateArdVtdu(ArdVtdu ardVtdu) {
- ardVtdu = mediaService.editPath(ardVtdu);
+ ardVtdu.setUpdateTime(DateUtils.getNowDate());
+ ardVtdu = ardMediaService.editPath(ardVtdu);
ardVtdu.setUpdateTime(DateUtils.getNowDate());
return ardVtduMapper.updateArdVtdu(ardVtdu);
}
@@ -104,7 +104,7 @@
for (String name : names) {
log.debug("流媒体【" + name + "】通道删除");
}
- mediaService.removePath(names);
+ ardMediaService.removePath(names);
return ardVtduMapper.deleteArdVtduByNames(names);
}
@@ -116,7 +116,7 @@
*/
@Override
public int deleteArdVtduByName(String name) {
- mediaService.removePath(name);
+ ardMediaService.removePath(name);
log.debug("流媒体【" + name + "】通道删除");
return ardVtduMapper.deleteArdVtduByName(name);
}
@@ -143,6 +143,9 @@
ardVtduMapper.clearAllArdVtdu();
}
+
+
+
//需要更新的（流媒体和vtdu相同）
@Override
public List<ArdVtdu> sameList(List<ArdVtdu> vtduses, List<String> names) {
@@ -152,18 +155,19 @@
//需要删除的（流媒体中多的）
@Override
public List<String> diffListToDel(List<ArdVtdu> vtduses, List<String> names) {
- if (vtduses.size() >= names.size()) {
- List<String> nameList = vtduses.stream().map(ArdVtdu::getName)
- .filter(item -> !names.contains(item))
- .collect(Collectors.toList());
- return nameList;
- } else {
- List<String> nameList =
- names.stream().filter(item -> !vtduses.stream().map(ArdVtdu::getName).collect(Collectors.toList()).contains(item))
- .collect(Collectors.toList());
- return nameList;
- }
+ // 先获取本地库非 publisher 的名称集合
+ Set<String> localNames = vtduses.stream()
+ .map(ArdVtdu::getName)
+ .collect(Collectors.toSet());
+
+ // 过滤 MediaMTX 流中非 publisher 流，并判断是否在本地库中
+ List<String> toDelete = names.stream()
+ .filter(n -> !localNames.contains(n)) // MediaMTX 有，但本地库没有
+ .collect(Collectors.toList());
+
+ return toDelete;
}
+
//需要新增的（流媒体中少的）
@Override
@@ -183,7 +187,7 @@
public void asyncMediaMTX(List<ArdVtdu> vtdus, List<String> names) {
//需要更新的数据,参数顺序注意
sameList(vtdus, names).forEach(vtdu -> {
- Conf conf = mediaService.getPathConfig(vtdu.getName());
+ Conf conf = ardMediaService.getPathConfig(vtdu.getName());
if (conf == null || conf.getSource() == null) {
return;
}
@@ -194,26 +198,108 @@
boolean isRtspSourceChanged = !Objects.equals(confRtspSource, currentRtspSource);
// 如果 RTSP 源发生变化，则更新路径
if (isRtspSourceChanged) {
- mediaService.editPath(vtdu);
+ ardMediaService.editPath(vtdu);
}
});
//需要删除的数据
diffListToDel(vtdus, names).forEach(name -> {
- mediaService.removePath(name);
+ ardMediaService.removePath(name);
});
//需要新增的数据
diffListToAdd(vtdus, names).forEach(vtdu -> {
- mediaService.addPath(vtdu);
+ ardMediaService.addPath(vtdu);
});
}
- //匹配rtsp地址
- private String matchRtsp(String value) {
- Matcher matcher = Pattern.compile("rtsp://[^\\s\"]+").matcher(value);
- if (matcher.find()) {
- return matcher.group();
- } else {
- return "";
+ /**
+ * 同步mediamtx
+ */
+ @Scheduled(cron = "0/10 * * * * ?")
+ public void syncMediamtx() {
+ log.debug("定时同步MediaMTX任务");
+ try {
+ List<String> nameList = ardMediaService.getNameList();
+ List<ArdVtdu> ardVtduList = this.selectArdVtduList(new ArdVtdu());
+ asyncMediaMTX(ardVtduList, nameList);
+ } catch (Exception e) {
+ log.error("同步MediaMTX任务失败", e);
+ }
+ }
+
+ /**
+ * 定时同步流媒体用户
+ */
+ @Scheduled(fixedRate = 5000)
+ public void syncStreamMedia() {
+
+ // Redis：在线 userId（1001）
+ Set<String> onlineUserIds = redisCache.getSetKey(CacheConstants.LOGIN_USERID_KEY)
+ .stream()
+ .map(k -> k.substring(CacheConstants.LOGIN_USERID_KEY.length()))
+ .collect(Collectors.toSet());
+
+ // DB：当前流媒体 userId（从 user_1001 提取 1001）
+ Set<String> currentStreamUserIds = ardVtduMapper
+ .selectArdVtduList(new ArdVtdu())
+ .stream()
+ .filter(v -> "publisher".equals(v.getRtspSource()))
+ .map(ArdVtdu::getName)
+ .filter(name -> name != null && name.startsWith("user_"))
+ .map(name -> name.substring("user_".length()))
+ .collect(Collectors.toSet());
+
+ // 1️⃣ 新增：Redis 有，DB 没有
+ onlineUserIds.stream()
+ .filter(id -> !currentStreamUserIds.contains(id))
+ .forEach(id -> {
+ ArdVtdu vtdu = ArdVtdu.builder()
+ .name("user_" + id)
+ .rtspSource("publisher")
+ .build();
+ insertArdVtdu(vtdu);
+ log.info("用户【{}】添加到流媒体", id);
+ });
+
+ // 2️⃣ 删除：DB 有，Redis 没有
+ currentStreamUserIds.stream()
+ .filter(id -> !onlineUserIds.contains(id))
+ .forEach(id -> {
+ deleteArdVtduByName("user_" + id);
+ log.info("用户【{}】移除流媒体", id);
+ });
+ }
+
+ @EventListener
+ public void handleLoginSuccess(AuthenticationSuccessEvent event) {
+ Object principal = event.getAuthentication().getPrincipal();
+ if (principal instanceof LoginUser) {
+ LoginUser loginUser = (LoginUser) principal; // 传统强制类型转换
+ Long userId = loginUser.getUser().getUserId();
+ String name="user_" + userId;
+ // 1️⃣ 判断数据库是否已存在
+ ArdVtdu existing = ardVtduMapper.selectArdVtduByName(name);
+ if (existing != null) {
+ log.info("流媒体【{}】已存在，跳过插入", name);
+ return;
+ }
+ ArdVtdu ardVtdu = ArdVtdu.builder()
+ .name(name) // 用 userId 作为流名称
+ .rtspSource("publisher") // 或 "publisher"，按业务区分
+ .build();
+ // 本地库插入
+ insertArdVtdu(ardVtdu);
+ log.info("用户【{}】登录，立即加入流媒体", name);
+ }
+ }
+ @EventListener
+ public void handleLogout(LogoutSuccessEvent event) {
+ Object principal = event.getAuthentication().getPrincipal();
+ if (principal instanceof LoginUser) {
+ LoginUser loginUser = (LoginUser) principal; // 传统强制类型转换
+ Long userId = loginUser.getUser().getUserId();
+ String name="user_" + userId;
+ deleteArdVtduByName(name);
+ log.info("用户【{}】注销，立即移除流媒体", name);
}
}
}
--
Gitblit v1.9.3