From a30a7797dfdad782050a20e7df975a227a8ddbba Mon Sep 17 00:00:00 2001
From: liusuyi <1951119284@qq.com>
Date: Mon, 04 Nov 2024 16:08:02 +0800
Subject: [PATCH] 优化
---
ard-work/pom.xml | 8 +
ard-work/src/main/java/com/ard/tianyu/websocket/utils/WebSocketUtil.java | 9 -
ard-work/src/main/java/com/ard/tianyu/websocket/config/WebSocketConfig.java | 9 ++
ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyStreamHandler.java | 123 ++++++++++++++++++++++++++++++
ard-work/src/main/java/com/ard/tianyu/websocket/model/PullStreamDTO.java | 27 ++++++
ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyMessageHandler.java | 12 +-
6 files changed, 172 insertions(+), 16 deletions(-)
diff --git a/ard-work/pom.xml b/ard-work/pom.xml
index 8506719..8bd2ef9 100644
--- a/ard-work/pom.xml
+++ b/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>
diff --git a/ard-work/src/main/java/com/ard/tianyu/websocket/config/WebSocketConfig.java b/ard-work/src/main/java/com/ard/tianyu/websocket/config/WebSocketConfig.java
index e43f086..f6579b0 100644
--- a/ard-work/src/main/java/com/ard/tianyu/websocket/config/WebSocketConfig.java
+++ b/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()); // 可选的拦截器
}
}
\ No newline at end of file
diff --git a/ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyMessageHandler.java b/ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyMessageHandler.java
index 58b5969..bb42f1f 100644
--- a/ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyMessageHandler.java
+++ b/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());
}
}
\ No newline at end of file
diff --git a/ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyStreamHandler.java b/ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyStreamHandler.java
new file mode 100644
index 0000000..09565c8
--- /dev/null
+++ b/ard-work/src/main/java/com/ard/tianyu/websocket/handler/TyStreamHandler.java
@@ -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());
+ }
+}
\ No newline at end of file
diff --git a/ard-work/src/main/java/com/ard/tianyu/websocket/model/PullStreamDTO.java b/ard-work/src/main/java/com/ard/tianyu/websocket/model/PullStreamDTO.java
new file mode 100644
index 0000000..3e1067d
--- /dev/null
+++ b/ard-work/src/main/java/com/ard/tianyu/websocket/model/PullStreamDTO.java
@@ -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;
+}
diff --git a/ard-work/src/main/java/com/ard/tianyu/websocket/utils/WebSocketUtil.java b/ard-work/src/main/java/com/ard/tianyu/websocket/utils/WebSocketUtil.java
index dc78204..8392fff 100644
--- a/ard-work/src/main/java/com/ard/tianyu/websocket/utils/WebSocketUtil.java
+++ b/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+$";
--
Gitblit v1.9.3