python读取监控流通过websocket连接发送到java服务端，服务端推流到前端

news/2024/5/9 14:45:06/文章来源:https://blog.csdn.net/weixin_44192389/article/details/132793858

在这里插入图片描述

python读取逐帧读取监控

import websocket
import base64
import cv2
import numpy as npvideoPath = "rtmp://ns8.indexforce.com/home/mystream"  // 此为公开RTSP流def on_message(ws, message):print(1)def connection_tmp(ws):websocket.enableTrace(True)ws = websocket.WebSocketApp("ws://127.0.0.1:8080/video",on_message=on_message,on_error=on_error,on_close=on_close)ws.on_open = on_opencap = cv2.VideoCapture(videoPath)try:ws.run_forever()except KeyboardInterrupt:ws.close()except:ws.close()def on_error(ws, error):reconnect_count=0print(type(error))print(error)print("正在尝试第%d次重连" % reconnect_count)reconnect_count += 1if reconnect_count < 100:connection_tmp(ws)def on_close(ws, close_status_code, close_msg):print("Connection closed:", close_status_code, close_msg)def on_open(ws):while cap.isOpened():ret, frame = cap.read()if not ret:breakencode_param = [int(cv2.IMWRITE_JPEG_QUALITY), 90]result, imgencode = cv2.imencode('.jpg', frame, encode_param)data = np.array(imgencode)img = data.tostring()img = base64.b64encode(img).decode()try:ws.send("img=" + img)except websocket.WebSocketConnectionClosedException:print("WebSocket connection is closed. Cannot send data.")breakwebsocket.enableTrace(True)
ws = websocket.WebSocketApp("ws://127.0.0.1:8080/video",on_message=on_message,on_error=on_error,on_close=on_close)ws.on_open = on_open
cap = cv2.VideoCapture(videoPath)
try:ws.run_forever()
except KeyboardInterrupt:ws.close()

Springboot集成websocket接收websocket连接请求
pom文件添加配置

  <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-websocket</artifactId><version>3.1.2</version></dependency><dependency><groupId>org.java-websocket</groupId><artifactId>Java-WebSocket</artifactId><version>1.3.5</version></dependency>

EnableWebsocket开启websocket

@SpringBootApplication(exclude={DataSourceAutoConfiguration.class,HibernateJpaAutoConfiguration.class})
@EnableWebSocket
public class MyblogApplication {public static void main(String[] args) {SpringApplication.run(MyblogApplication.class, args);}}

package sxu.edu.hkj.myblog.config;//package sxu.edu.hkj.myblog.config;import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry;
import org.springframework.web.socket.server.standard.ServletServerContainerFactoryBean;
import sxu.edu.hkj.myblog.controller.VideoWebSocketHandler;@Configuration
@EnableWebSocket
public class WebSocketConfig implements WebSocketConfigurer {@Overridepublic void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {// 在这里配置WebSocket处理程序并指定自定义端点路径registry.addHandler(new VideoWebSocketHandler(), "/video").setAllowedOrigins("*");}@Beanpublic ServletServerContainerFactoryBean createWebSocketContainer() {ServletServerContainerFactoryBean container = new ServletServerContainerFactoryBean();// 在此处设置bufferSizecontainer.setMaxTextMessageBufferSize(1024000);container.setMaxBinaryMessageBufferSize(1024000);container.setMaxSessionIdleTimeout(15 * 60000L);return container;}}

通信服务：

package sxu.edu.hkj.myblog.controller;
import lombok.SneakyThrows;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.BinaryMessage;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.BinaryWebSocketHandler;import java.io.IOException;
import java.util.concurrent.CopyOnWriteArraySet;
import java.nio.ByteBuffer;@Component
public class VideoWebSocketHandler extends BinaryWebSocketHandler {private static final CopyOnWriteArraySet<WebSocketSession> SESSIONS = new CopyOnWriteArraySet<>();@Overridepublic void afterConnectionEstablished(WebSocketSession session) {SESSIONS.add(session);System.out.println("WebSocket消息】有新的连接，总数为：" + SESSIONS.size());}@Overridepublic void afterConnectionClosed(WebSocketSession session, CloseStatus status) {SESSIONS.remove(session);System.out.println("【WebSocket消息】连接断开，总数为：" + SESSIONS.size());}@SneakyThrows@Overrideprotected void handleTextMessage(WebSocketSession session, TextMessage message){// Handle incoming text messages here, if neededString textMessage = message.getPayload();
//        System.out.println("Received text message: " + textMessage);// Broadcast the received text message to all connected clientsfor (WebSocketSession clientSession : SESSIONS) {if (clientSession.isOpen() && !clientSession.equals(session)) {clientSession.sendMessage(new TextMessage(textMessage));}}}@Overrideprotected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) throws IOException {byte[] frameData = message.getPayload().array();System.out.println(frameData);// 将接收到的二进制图像数据发送给所有客户端for (WebSocketSession clientSession : SESSIONS) {try {if (clientSession.isOpen() && !clientSession.equals(session)) {// 创建一个新的BinaryMessage并发送给客户端BinaryMessage responseMessage = new BinaryMessage(ByteBuffer.wrap(frameData));clientSession.sendMessage(responseMessage);System.out.println("发送关闭广播++++++++++++++++++++++"+frameData);}} catch (IOException e) {e.printStackTrace();}}}
}

前端接收websocket广播信息，并绑定dom到前端

<imgclass="resizable-image11":src="image3Source"alt="Image"v-show="isCalibrating"style="opacity: 0.6; height: 560px; width: 1168px; margin:-569px -29px 12px -30.06px; pointer-events: none; filter: contrast(550%)"/>let ws = new WebSocket("ws://127.0.0.1:8082/video");ws.onopen = function(evt) {
};
ws.onopen = function (evt) {ws.send("Hello WebSockets!");isWebSocketConnected.value = true; // WebSocket is open
};ws.onclose = function (evt) {isWebSocketConnected.value = false; // WebSocket is closedvar reconnect_count=0reconnect_count += 1if (reconnect_count < 100){let ws = new WebSocket("ws://127.0.0.1:8082/video");}};ws.onmessage = function (event) {const message = event.data;const parts = message.split('=');const dataType = parts[0];const data = parts[1];if (dataType === 'img') {const imageData = 'data:image/jpeg;base64,' + data;document.getElementById('resImg').src = imageData;} else if (dataType === 'a1') {}
};ws.onclose = function(evt) {
};