优化

liusuyi

2024-11-04 a30a7797dfdad782050a20e7df975a227a8ddbba

优化

 ard-work/pom.xml

@@ -23,10 +23,11 @@
            <artifactId>ard-common</artifactId>
        </dependency>

        <!--knife4j依赖-->
        <!--引入Knife4j的官方start包,该指南选择Spring Boot版本<3.0,开发者需要注意-->
        <dependency>
            <groupId>com.github.xiaoymin</groupId>
            <artifactId>knife4j-spring-boot-starter</artifactId>
            <artifactId>knife4j-openapi2-spring-boot-starter</artifactId>
            <version>4.4.0</version>
        </dependency>

        <!-- Forest -->
@@ -49,13 +50,14 @@
            <version>1.18.28</version>
        </dependency>

        <!--protobuf的依赖-->
        <!--protobuf依赖-->
        <dependency>
            <groupId>com.google.protobuf</groupId>
            <artifactId>protobuf-java</artifactId>
            <version>3.20.1</version>
        </dependency>

        <!--websocket依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>

 ard-work/src/main/java/com/ard/tianyu/websocket/config/WebSocketConfig.java

@@ -1,6 +1,7 @@
package com.ard.tianyu.websocket.config;

import com.ard.tianyu.websocket.handler.TyMessageHandler;
import com.ard.tianyu.websocket.handler.TyStreamHandler;
import com.ard.tianyu.websocket.interceptor.TyHandshakeInterceptor;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
@@ -14,14 +15,20 @@

    @Resource
    private TyMessageHandler tyMessageHandler;
    @Resource
    private TyStreamHandler tyStreamHandler;

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        //添加myWebSocketHandler消息处理对象以及websocket连接地址
        registry.addHandler(tyMessageHandler, "/ws/**")
        registry.addHandler(tyMessageHandler, "/ws/state/**")
                //设置允许跨域访问
                .setAllowedOrigins("*")
                //添加拦截器可实现用户链接前进行权限校验等操作
                .addInterceptors(new TyHandshakeInterceptor());
        // 注册流媒体 WebSocket 处理器
        registry.addHandler(tyStreamHandler, "/ws/stream/**")
                .setAllowedOrigins("*")
                .addInterceptors(new TyHandshakeInterceptor()); // 可选的拦截器
    }
}

 ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyMessageHandler.java

@@ -16,24 +16,24 @@
    public void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        // 处理接收到的消息
        String payload = message.getPayload();
        log.info("【websocket】Received message: " + payload);
        log.info("【websocket】size: " + WebSocketUtil.ONLINE_USER_SESSIONS.size());
        log.info("【cmd】Received message: " + payload);
        log.info("【cmd】size: " + WebSocketUtil.ONLINE_USER_SESSIONS.size());
    }

    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        log.info("【websocket】Connected: " + session.getId());
        log.info("【cmd】Connected: " + session.getId());
        String userId = (String) session.getAttributes().get("userId");
        WebSocketUtil.ONLINE_USER_SESSIONS.put(userId, session);
        log.info("【websocket】size: " + WebSocketUtil.ONLINE_USER_SESSIONS.size());
        log.info("【cmd】size: " + WebSocketUtil.ONLINE_USER_SESSIONS.size());

    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        log.info("【websocket】Disconnected: " + session.getId());
        log.info("【cmd】Disconnected: " + session.getId());
        String userId = (String) session.getAttributes().get("userId");
        WebSocketUtil.ONLINE_USER_SESSIONS.remove(userId);
        log.info("【websocket】size: " + WebSocketUtil.ONLINE_USER_SESSIONS.size());
        log.info("【cmd】size: " + WebSocketUtil.ONLINE_USER_SESSIONS.size());
    }
}

 ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyStreamHandler.java

New file
@@ -0,0 +1,123 @@
package com.ard.tianyu.websocket.handler;

import com.alibaba.fastjson2.JSON;
import com.ard.tianyu.model.receiver.BoxStreamUrlReceiver;
import com.ard.tianyu.service.ITyService;
import com.ard.tianyu.websocket.model.PullStreamDTO;
import com.ard.tianyu.websocket.utils.WebSocketUtil;
import com.ard.vtdu.domain.ArdVtdu;
import com.ard.vtdu.service.IMediaService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;

import javax.annotation.Resource;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

@Component
@Slf4j
public class TyStreamHandler extends TextWebSocketHandler {

    @Resource
    private ITyService tyService;
    @Resource
    private IMediaService mediaService;

    //存储盒子的拉流地址
    private static final ConcurrentMap<String, String> BOX_STREAM_PULL_CACHE = new ConcurrentHashMap<>();
    //存储盒子的拉流用户集合
    private static final ConcurrentMap<String, Set<String>> STREAM_USER_MAP = new ConcurrentHashMap<>();

    @Override
    public void handleTextMessage(WebSocketSession session, TextMessage message) {
        String userId = (String) session.getAttributes().get("userId");
        String payload = message.getPayload();
        PullStreamDTO pullStreamDTO = JSON.parseObject(payload, PullStreamDTO.class);

        if (pullStreamDTO != null) {
            String boxSn = pullStreamDTO.getBoxSn();
            if (pullStreamDTO.getPullType().equals(1)) {
                startPullStream(boxSn, userId);
            } else {
                stopPullStream(boxSn, userId);
            }
        }

        log.info("【stream】Received message: " + payload);
        log.info("【stream】size: " + WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.size());
    }

    //开始拉流
    private void startPullStream(String boxSn, String userId) {
        //BOX_STREAM_USER_COUNT.merge(boxSn, 1, Integer::sum);
        STREAM_USER_MAP.computeIfAbsent(boxSn, k -> ConcurrentHashMap.newKeySet()).add(userId);
        String messageReply;
        if (BOX_STREAM_PULL_CACHE.containsKey(boxSn)) {
            String webrtcUrl = BOX_STREAM_PULL_CACHE.get(boxSn);
            messageReply = webrtcUrl;
            log.warn("【stream】Already pulled stream: " + webrtcUrl);
        } else {
            BoxStreamUrlReceiver boxLiveUrl = tyService.getBoxLiveUrl(boxSn);
            if (boxLiveUrl != null) {
                ArdVtdu vtdu = ArdVtdu.builder().name(boxSn).mode(0).isCode(0).rtspSource(boxLiveUrl.getRtmp()).build();
                ArdVtdu ardVtdu = mediaService.addPath(vtdu);
                BOX_STREAM_PULL_CACHE.put(boxSn, ardVtdu.getWebrtcUrl());
                messageReply = ardVtdu.getWebrtcUrl();
                log.warn("【stream】First pulled stream: " + ardVtdu.getWebrtcUrl());
            } else {
                log.warn("【stream】Box stream not found: " + boxSn);
                messageReply = "【stream】Box stream not found: " + boxSn;
            }
        }
        WebSocketUtil.sendMessage(WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.get(userId), messageReply);
    }

    //停止拉流
    private void stopPullStream(String boxSn, String userId) {
        synchronized (this) {
            Set<String> users = STREAM_USER_MAP.get(boxSn);
            if (users != null) {
                users.remove(userId);
                int userCount = users.size();
                String messageReply;
                if (userCount == 0) {
                    closeStream(boxSn);
                    messageReply = "【stream】Stream closed due to no active users: " + boxSn;
                } else {
                    messageReply = "【stream】Stream is still in use, remaining users: " + users;
                }
                WebSocketUtil.sendMessage(WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.get(userId), messageReply);
            }
        }
    }

    //关闭拉流
    private void closeStream(String boxSn) {
        mediaService.removePath(boxSn);
        BOX_STREAM_PULL_CACHE.remove(boxSn);
        STREAM_USER_MAP.remove(boxSn);
        log.warn("【stream】Stream closed due to no active users: " + boxSn);
    }

    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        String userId = (String) session.getAttributes().get("userId");
        WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.put(userId, session);
        log.info("【stream】Connected: " + session.getId());
        log.info("【stream】size: " + WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.size());
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        String userId = (String) session.getAttributes().get("userId");
        WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.remove(userId);
        STREAM_USER_MAP.forEach((boxSn, users) -> stopPullStream(boxSn, userId));
        log.info("【stream】Disconnected: " + session.getId());
        log.info("【stream】size: " + WebSocketUtil.STREAM_ONLINE_USER_SESSIONS.size());
    }
}

 ard-work/src/main/java/com/ard/tianyu/websocket/model/PullStreamDTO.java

New file
@@ -0,0 +1,27 @@
package com.ard.tianyu.websocket.model;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * PullStreamDTO
 *
 * @author liusuyi
 * @date 2024年11月04日14:40
 **/
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class PullStreamDTO {
    /**
     * 盒子编号
     */
    private String boxSn;
    /**
     * 类型 0-未拉流 1-拉流
     */
    private Integer pullType;
}

 ard-work/src/main/java/com/ard/tianyu/websocket/utils/WebSocketUtil.java

@@ -27,6 +27,8 @@

    // 存储 websocket session
    public static final ConcurrentMap<String, WebSocketSession> ONLINE_USER_SESSIONS = new ConcurrentHashMap<>();
    // 存储 拉流 websocket session
    public static final ConcurrentMap<String, WebSocketSession> STREAM_ONLINE_USER_SESSIONS = new ConcurrentHashMap<>();

    /**
     * @param session 用户 session
@@ -96,7 +98,6 @@
    public static void sendMessageAll(String message) {
        ONLINE_USER_SESSIONS.forEach((sessionId, session) -> sendMessage(session, message));
    }

    /**
     * 推送消息到其他客户端
     *
@@ -124,11 +125,7 @@
        ONLINE_USER_SESSIONS.forEach((sessionId, session) -> sendMessage(session, JSON.toJSONString(message)));
    }
    /**
     * @Author 刘苏义
     * @Description  发送消息给userId_前缀的人
     * @Date   2024/7/16 10:24
     * @Param
     * @return
     * 发送消息给userId_前缀的人
     */
    public static void sendMessagePrefix(String targetId, String message) {
        String regex = "^" + Pattern.quote(targetId) + "_\\d+$";