[关闭]
@liruiyi962464 2025-07-26T07:31:41.000000Z 字数 8824 阅读 74

Python实时传送自建视频流,Java WebSocket接收

代码

测试使用版本：jeecg-boot 3.7.2

Python Dome

  1. import asyncio
  2. import cv2
  3. import numpy as np
  4. import mss
  5. import time
  6. import websockets
  8. # WebSocket 服务器地址
  9. URI = "ws://192.168.10.167:8080/jeecg-boot/video-stream"
  11. async def send_frame(websocket):
  12.     with mss.mss() as sct:
  13.         monitor = sct.monitors[1]
  14.         width, height = int(monitor["width"] * 0.5), int(monitor["height"] * 0.5)
  16.         # 使用 H.264 编码器（需安装支持的 OpenCV）
  17.         # 如果报错，尝试 cv2.VideoWriter_fourcc(*'XVID')
  18.         fourcc = cv2.VideoWriter_fourcc(*'X264')  # 或 'H264', 'XVID'
  19.         fps = 5
  20.         frame_interval = 1.0 / fps
  21.         last_time = time.time()
  23.         print("开始实时录屏并上传...")
  25.         while True:
  26.             current_time = time.time()
  27.             if current_time - last_time < frame_interval:
  28.                 continue
  30.             # 截图并处理
  31.             screenshot = sct.grab(monitor)
  32.             frame = np.array(screenshot)[:, :, :3]  # 去掉 alpha 通道
  33.             frame = cv2.resize(frame, (width, height))
  35.             # 编码为 H.264 帧
  36.             ret, buffer = cv2.imencode('.jpg', frame, [int(cv2.IMWRITE_JPEG_QUALITY), 70])
  37.             if not ret:
  38.                 continue
  40.             await websocket.send(buffer.tobytes())
  41.             last_time = current_time
  43. async def main():
  44.     async with websockets.connect(URI) as websocket:
  45.         await send_frame(websocket)
  47. if __name__ == "__main__":
  48.     asyncio.run(main())

ShiroConfig.java

放开接口，因为没有token,ShiroConfig增加下列代码

  1. filterChainDefinitionMap.put("/video-stream", "anon"); // 放行 视频播放路径

WebSocketConfig.java

增加注解@EnableWebSocket
WebSocketConfig实现WebSocketConfigurer接口

  1. package org.jeecg.config;
  3. import org.jeecg.VideoStreamHandler;
  4. import org.jeecg.config.filter.WebsocketFilter;
  5. import org.springframework.beans.factory.annotation.Autowired;
  6. import org.springframework.boot.web.servlet.FilterRegistrationBean;
  7. import org.springframework.context.annotation.Bean;
  8. import org.springframework.context.annotation.Configuration;
  9. import org.springframework.web.socket.config.annotation.EnableWebSocket;
  10. import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
  11. import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
  12. import org.springframework.web.socket.server.standard.ServerEndpointExporter;
  14. /**
  15.  * @Description: WebSocketConfig
  16.  * @author: jeecg-boot
  17.  */
  18. @Configuration
  19. @EnableWebSocket
  20. public class WebSocketConfig implements WebSocketConfigurer {
  21.     /**
  22.      *  注入ServerEndpointExporter，
  23.      *  这个bean会自动注册使用了@ServerEndpoint注解声明的Websocket endpoint
  24.      */
  25.     @Bean
  26.     public ServerEndpointExporter serverEndpointExporter() {
  27.         return new ServerEndpointExporter();
  28.     }
  30.     @Bean
  31.     public WebsocketFilter websocketFilter(){
  32.         return new WebsocketFilter();
  33.     }
  36.     @Bean
  37.     public FilterRegistrationBean getFilterRegistrationBean() {
  38.         FilterRegistrationBean bean = new FilterRegistrationBean();
  39.         bean.setFilter(websocketFilter());
  40.         bean.addUrlPatterns(
  41.                 "/taskCountSocket/*",
  42.                 "/websocket/*",
  43.                 "/eoaSocket/*",
  44.                 "/eoaNewChatSocket/*",
  45.                 "/newsWebsocket/*",
  46.                 "/dragChannelSocket/*",
  47.                 "/vxeSocket/*"
  48.         );
  49.         return bean;
  50.     }
  52.     // 以下为新增代码
  53.     @Autowired
  54.     private VideoStreamHandler videoStreamHandler;
  56.     // 实现的接口
  57.     @Override
  58.     public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
  59.         System.out.println("======== 正在注册 WebSocket 端点: /video-stream ========");
  60.         registry.addHandler(videoStreamHandler, "/video-stream")
  61.                 .setAllowedOrigins("*");
  62.     }
  63. }

VideoStreamMessage.java

  1. package org.jeecg;
  3. import lombok.Data;
  5. /**
  6.  * 视频流消息实体类
  7.  */
  8. @Data
  9. public class VideoStreamMessage {
  10.     private String type;     // 消息类型："params"（参数消息）或 "data"（数据消息）或 "stop"（停止消息）
  11.     private String fileName; // 视频文件名（如未提供，服务端自动生成）
  12.     private String sessionId; // 会话ID（通常由WebSocket自动生成）
  13.     private String from;     // 发送方标识（如设备ID、客户端名称）
  14.     private String content;  // 消息内容（参数消息中为元数据，数据消息中为描述）
  15.     private long timestamp;  // 时间戳（毫秒级，用于同步或日志）
  16. }

VideoStreamHandler.java

实现接受视频流
支持接收参数

  1. package org.jeecg;
  3. import com.fasterxml.jackson.databind.ObjectMapper;
  4. import lombok.Data;
  5. import org.slf4j.Logger;
  6. import org.slf4j.LoggerFactory;
  7. import org.springframework.stereotype.Component;
  8. import org.springframework.web.socket.BinaryMessage;
  9. import org.springframework.web.socket.TextMessage;
  10. import org.springframework.web.socket.WebSocketSession;
  11. import org.springframework.web.socket.handler.BinaryWebSocketHandler;
  13. import java.io.FileOutputStream;
  14. import java.io.IOException;
  15. import java.nio.ByteBuffer;
  16. import java.text.SimpleDateFormat;
  17. import java.util.Date;
  18. import java.util.Map;
  19. import java.util.concurrent.ConcurrentHashMap;
  20. import java.util.concurrent.atomic.AtomicBoolean;
  22. /**
  23.  * 视频流WebSocket处理器
  24.  */
  25. @Component
  26. public class VideoStreamHandler extends BinaryWebSocketHandler {
  27.     private static final Logger logger = LoggerFactory.getLogger(VideoStreamHandler.class);
  28.     private static final String OUTPUT_DIR = "D:\\opt\\upFiles\\"; // 视频文件输出目录
  29.     private final ObjectMapper objectMapper = new ObjectMapper(); // JSON解析器
  31.     // 存储会话状态
  32.     private final Map<String, FileOutputStream> sessionOutputs = new ConcurrentHashMap<>(); // 文件输出流
  33.     private final Map<String, VideoStreamParams> sessionParams = new ConcurrentHashMap<>();   // 会话参数
  34.     private final Map<String, AtomicBoolean> writingFlags = new ConcurrentHashMap<>();      // 写入状态标志
  36.     /**
  37.      * 会话参数内部类
  38.      */
  39.     @Data
  40.     private static class VideoStreamParams {
  41.         private String fileName; // 文件名
  42.         private String from;     // 发送方
  43.         private String content;  // 内容描述
  44.         private long timestamp;  // 时间戳
  45.     }
  47.     @Override
  48.     public void afterConnectionEstablished(WebSocketSession session) throws Exception {
  49.         logger.info("===== 新WebSocket连接 =====");
  50.         logger.info("Session ID: {}", session.getId());
  51.         logger.info("URI: {}", session.getUri());
  52.         writingFlags.put(session.getId(), new AtomicBoolean(true)); // 初始化写入标志
  53.     }
  55.     @Override
  56.     protected void handleTextMessage(WebSocketSession session, TextMessage message) {
  57.         String payload = message.getPayload();
  58.         logger.info("收到文本消息: {}", payload);
  60.         try {
  61.             VideoStreamMessage msg = objectMapper.readValue(payload, VideoStreamMessage.class);
  63.             if ("params".equals(msg.getType())) {
  64.                 // 处理参数消息：初始化会话参数和文件输出流
  65.                 VideoStreamParams params = new VideoStreamParams();
  66.                 params.setFileName(msg.getFileName());
  67.                 params.setFrom(msg.getFrom());
  68.                 params.setContent(msg.getContent());
  69.                 params.setTimestamp(msg.getTimestamp());
  70.                 sessionParams.put(session.getId(), params);
  72.                 // === 新增：输出解析后的参数 ===
  73.                 logger.info("===== 视频流参数 =====");
  74.                 logger.info("文件名: {}", msg.getFileName());
  75.                 logger.info("来源: {}", msg.getFrom());
  76.                 logger.info("内容: {}", msg.getContent());
  77.                 logger.info("时间戳: {}", msg.getTimestamp());
  78.                 SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
  79.                 String formattedTime = sdf.format(new Date(msg.getTimestamp()));
  80.                 logger.info("格式化时间: {}", formattedTime);
  82.                 // 创建输出文件（使用会话ID作为默认文件名）
  83.                 String outputFile = OUTPUT_DIR + (msg.getFileName() != null ?
  84.                         msg.getFileName() : "video_" + session.getId() + ".mp4");
  85.                 sessionOutputs.put(session.getId(), new FileOutputStream(outputFile));
  86.                 logger.info("视频流保存路径: {}", outputFile);
  88.             } else if ("stop".equalsIgnoreCase(msg.getType())) {
  89.                 // 处理停止消息
  90.                 closeSession(session);
  91.             }
  92.         } catch (Exception e) {
  93.             logger.warn("文本消息解析失败: {}", e.getMessage());
  94.         }
  95.     }
  97.     @Override
  98.     protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) {
  99.         // 检查会话是否有效
  100.         AtomicBoolean flag = writingFlags.get(session.getId());
  101.         if (flag == null || !flag.get()) return;
  103.         // 写入二进制数据到文件
  104.         FileOutputStream fos = sessionOutputs.get(session.getId());
  105.         if (fos != null) {
  106.             try {
  107.                 ByteBuffer buffer = message.getPayload();
  108.                 fos.write(buffer.array());
  109.             } catch (IOException e) {
  110.                 logger.error("二进制写入失败: {}", e.getMessage());
  111.                 closeSession(session);
  112.             }
  113.         }
  114.     }
  116.     @Override
  117.     public void afterConnectionClosed(WebSocketSession session, org.springframework.web.socket.CloseStatus status) {
  118.         closeSession(session); // 清理资源
  119.     }
  121.     /**
  122.      * 关闭会话并释放资源
  123.      */
  124.     private void closeSession(WebSocketSession session) {
  125.         String sessionId = session.getId();
  126.         AtomicBoolean flag = writingFlags.getOrDefault(sessionId, new AtomicBoolean(false));
  127.         flag.set(false); // 禁止继续写入
  129.         try {
  130.             FileOutputStream fos = sessionOutputs.get(sessionId);
  131.             if (fos != null) {
  132.                 fos.close();
  133.                 logger.info("会话结束，文件已保存: {}", sessionParams.get(sessionId).getFileName());
  134.             }
  135.         } catch (IOException e) {
  136.             logger.error("文件关闭失败: {}", e.getMessage());
  137.         } finally {
  138.             sessionOutputs.remove(sessionId);
  139.             sessionParams.remove(sessionId);
  140.             writingFlags.remove(sessionId);
  141.         }
  142.     }
  143. }

测试

开始推送

  1. {
  2.     "type": "params",
  3.     "fileName": "test_video.mp4",
  4.     "from": "camera_01",
  5.     "content": "分辨率=1080p, 编码格式=h264, 帧率=30fps",
  6.     "timestamp": 1630000000000
  7. }

结束推送

  1. {
  2.     "type": "stop"
  3. }

结果

  1. 2025-07-25 17:45:55.356 [XNIO-1 task-2] INFO  org.jeecg.VideoStreamHandler:49 - ===== 新WebSocket连接 =====
  2. 2025-07-25 17:45:55.357 [XNIO-1 task-2] INFO  org.jeecg.VideoStreamHandler:50 - Session ID: b5391ec8-5bf2-b9d3-8def-8d7ae0668940
  3. 2025-07-25 17:45:55.357 [XNIO-1 task-2] INFO  org.jeecg.VideoStreamHandler:51 - URI: ws://192.168.10.167:8080/jeecg-boot/video-stream
  4. 2025-07-25 17:46:00.642 [XNIO-1 I/O-11] INFO  org.jeecg.VideoStreamHandler:58 - 收到文本消息: {
  5.     "type": "params",
  6.     "fileName": "test_video.mp4",
  7.     "from": "camera_01",
  8.     "content": "分辨率=1080p, 编码格式=h264, 帧率=30fps",
  9.     "timestamp": 1630000000000
  10. }
  11. 2025-07-25 17:46:00.642 [XNIO-1 I/O-11] INFO  org.jeecg.VideoStreamHandler:73 - ===== 视频流参数 =====
  12. 2025-07-25 17:46:00.642 [XNIO-1 I/O-11] INFO  org.jeecg.VideoStreamHandler:74 - 文件名: test_video.mp4
  13. 2025-07-25 17:46:00.642 [XNIO-1 I/O-11] INFO  org.jeecg.VideoStreamHandler:75 - 来源: camera_01
  14. 2025-07-25 17:46:00.642 [XNIO-1 I/O-11] INFO  org.jeecg.VideoStreamHandler:76 - 内容: 分辨率=1080p, 编码格式=h264, 帧率=30fps
  15. 2025-07-25 17:46:00.642 [XNIO-1 I/O-11] INFO  org.jeecg.VideoStreamHandler:77 - 时间戳: 1630000000000
  16. 2025-07-25 17:46:00.643 [XNIO-1 I/O-11] INFO  org.jeecg.VideoStreamHandler:80 - 格式化时间: 2021-08-27 01:46:40
  17. 2025-07-25 17:46:00.643 [XNIO-1 I/O-11] INFO  org.jeecg.VideoStreamHandler:86 - 视频流保存路径: D:\opt\upFiles\test_video.mp4
  18. 2025-07-25 17:46:13.486 [XNIO-1 I/O-11] INFO  org.jeecg.VideoStreamHandler:58 - 收到文本消息: {
  19.     "type": "stop"
  20. }
  21. 2025-07-25 17:46:15.381 [XNIO-1 I/O-11] INFO  org.jeecg.VideoStreamHandler:133 - 会话结束，文件已保存: test_video.mp4
添加新批注
在作者公开此批注前，只有你和作者可见。
回复批注