优化

liusuyi

2026-04-01 a2bef1bc001cc196affb1b24d69eec4127cfa40a

 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);
        }
    }
}