From 67fdf3ec49bff5ca53b299eec9d749cc80f1ea18 Mon Sep 17 00:00:00 2001
From: liusuyi <1951119284@qq.com>
Date: Tue, 20 Jan 2026 14:14:05 +0800
Subject: [PATCH] 用户添加流媒体
---
ard-common/src/main/java/com/ard/common/core/redis/RedisCache.java | 18 ++++-
ard-framework/src/main/java/com/ard/framework/security/handle/CustomAuthenticationSuccessHandler.java | 18 +++++
ard-framework/src/main/java/com/ard/framework/web/service/SysLoginService.java | 6 -
ard-vtdu/src/main/java/com/ard/vtdu/service/impl/ArdVtduServiceImpl.java | 100 +++++++++++++++++++++++++++-----
ard-work/src/main/java/com/ard/work/service/task/MediaTask.java | 4
5 files changed, 116 insertions(+), 30 deletions(-)
diff --git a/ard-common/src/main/java/com/ard/common/core/redis/RedisCache.java b/ard-common/src/main/java/com/ard/common/core/redis/RedisCache.java
index ef44326..09661f1 100644
--- a/ard-common/src/main/java/com/ard/common/core/redis/RedisCache.java
+++ b/ard-common/src/main/java/com/ard/common/core/redis/RedisCache.java
@@ -1,11 +1,9 @@
package com.ard.common.core.redis;
-import java.util.Collection;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;
import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundSetOperations;
import org.springframework.data.redis.core.HashOperations;
@@ -274,4 +272,14 @@
List<Object> values = redisTemplate.opsForValue().multiGet(keys);
return values;
}
+ /**
+ * 根据前缀获取所有的key* 例如：pro_
+ */
+ public Set<String> getSetKey(String prefix) {
+ Set<String> keys = redisTemplate.keys(prefix + "*");
+ if (keys == null) return Collections.emptySet();
+ return keys.stream().map(Object::toString).collect(Collectors.toSet());
+ }
+
+
}
diff --git a/ard-framework/src/main/java/com/ard/framework/security/handle/CustomAuthenticationSuccessHandler.java b/ard-framework/src/main/java/com/ard/framework/security/handle/CustomAuthenticationSuccessHandler.java
index a660bda..1efc113 100644
--- a/ard-framework/src/main/java/com/ard/framework/security/handle/CustomAuthenticationSuccessHandler.java
+++ b/ard-framework/src/main/java/com/ard/framework/security/handle/CustomAuthenticationSuccessHandler.java
@@ -2,9 +2,14 @@
import com.ard.common.constant.CacheConstants;
import com.ard.common.constant.Constants;
+import com.ard.common.core.domain.model.LoginUser;
import com.ard.common.core.redis.RedisCache;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.security.authentication.event.AuthenticationSuccessEvent;
+import org.springframework.security.core.Authentication;
+import org.springframework.security.core.context.SecurityContextHolder;
import org.springframework.stereotype.Component;
@Component
@@ -16,15 +21,26 @@
@Autowired
private RedisCache redisCache;
- public void onAuthenticationSuccess(Long userId) {
+ @Autowired
+ private ApplicationEventPublisher publisher;
+
+ public void onAuthenticationSuccess(Authentication authentication) {
// 允许多端登录，直接放行
if (soloLogin) {
return;
}
// 检查 Redis 中是否已有该用户登录
+ LoginUser loginUser = (LoginUser) authentication.getPrincipal();
+ Long userId = loginUser.getUser().getUserId();
+
String key = CacheConstants.LOGIN_USERID_KEY + userId;
if (redisCache.hasKey(key)) {
throw new RuntimeException("用户已登录，请勿重复登录");
}
+ // 登录成功事件推送
+ if (authentication != null && authentication.isAuthenticated()) {
+ publisher.publishEvent(new AuthenticationSuccessEvent(authentication));
+ }
}
+
}
diff --git a/ard-framework/src/main/java/com/ard/framework/web/service/SysLoginService.java b/ard-framework/src/main/java/com/ard/framework/web/service/SysLoginService.java
index 1d4773c..3795c79 100644
--- a/ard-framework/src/main/java/com/ard/framework/web/service/SysLoginService.java
+++ b/ard-framework/src/main/java/com/ard/framework/web/service/SysLoginService.java
@@ -89,14 +89,10 @@
AuthenticationContextHolder.setContext(authenticationToken);
// 该方法会去调用UserDetailsServiceImpl.loadUserByUsername
authentication = authenticationManager.authenticate(authenticationToken);
-
- // 获取用户ID
- LoginUser loginUser = (LoginUser) userDetailsService.loadUserByUsername(username);
- Long userId = loginUser.getUser().getUserId();
// 登录成功逻辑
if (authentication.isAuthenticated()) {
// 检查是否已登录
- successHandler.onAuthenticationSuccess(userId);
+ successHandler.onAuthenticationSuccess(authentication);
}
}
catch (Exception e)
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 139c5cc..f144994 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,15 @@
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.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;
/**
@@ -29,8 +33,8 @@
private ArdVtduMapper ardVtduMapper;
@Resource
private IArdMediaService ardMediaService;
-
-
+ @Resource
+ private RedisCache redisCache;
@PostConstruct
public void initMediaMTX() {
@@ -146,18 +150,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
@@ -211,9 +216,70 @@
List<String> nameList = ardMediaService.getNameList();
List<ArdVtdu> ardVtduList = this.selectArdVtduList(new ArdVtdu());
asyncMediaMTX(ardVtduList, nameList);
- }
- catch (Exception e) {
+ } 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 username = loginUser.getUsername();
+ ArdVtdu ardVtdu = ArdVtdu.builder()
+ .name("user_" + userId) // 用 userId 作为流名称
+ .rtspSource("publisher") // 或 "publisher"，按业务区分
+ .build();
+ // 本地库插入
+ insertArdVtdu(ardVtdu);
+ log.info("用户【{}】登录，立即加入流媒体", username);
+ } else {
+ log.warn("登录事件 principal 不是 LoginUser，无法获取 userId");
+ }
+ }
}
diff --git a/ard-work/src/main/java/com/ard/work/service/task/MediaTask.java b/ard-work/src/main/java/com/ard/work/service/task/MediaTask.java
index 264e117..fe22181 100644
--- a/ard-work/src/main/java/com/ard/work/service/task/MediaTask.java
+++ b/ard-work/src/main/java/com/ard/work/service/task/MediaTask.java
@@ -57,9 +57,9 @@
private final AtomicBoolean runningPic = new AtomicBoolean(false);
@Scheduled(cron = "0/10 * * * * ?")
public void syncVideoMediaUrl() {
- log.warn("开始同步视频媒体");
+ log.debug("开始同步视频媒体");
if (!runningVideo.compareAndSet(false, true)) {
- log.warn("上一次同步还没完成，跳过本次调度");
+ log.debug("上一次同步还没完成，跳过本次调度");
return;
}
try {
--
Gitblit v1.9.3