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

• python读取逐帧读取监控

import websocket
import base64
import cv2
import numpy as np

videoPath = "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_open
    cap = cv2.VideoCapture(videoPath)
    try:
        ws.run_forever()
    except KeyboardInterrupt:
        ws.close()
    except:
        ws.close()

def on_error(ws, error):
    reconnect_count=0
    print(type(error))
    print(error)
    print("正在尝试第%d次重连" % reconnect_count)
    reconnect_count += 1
    if 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:
            break

        encode_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.")
            break



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_open
cap = cv2.VideoCapture(videoPath)
try:
    ws.run_forever()
except KeyboardInterrupt:
    ws.close()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

• 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>
1
2
3
4
5
6
7
8
9
10

EnableWebsocket开启websocket

@SpringBootApplication(exclude={DataSourceAutoConfiguration.class,HibernateJpaAutoConfiguration.class})
@EnableWebSocket
public class MyblogApplication {

    public static void main(String[] args) {
        SpringApplication.run(MyblogApplication.class, args);
    }

}

1
2
3
4
5
6
7
8
9
10

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 {
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        // 在这里配置WebSocket处理程序并指定自定义端点路径
        registry.addHandler(new VideoWebSocketHandler(), "/video").setAllowedOrigins("*");
    }

    @Bean
    public ServletServerContainerFactoryBean createWebSocketContainer() {
        ServletServerContainerFactoryBean container = new ServletServerContainerFactoryBean();
        // 在此处设置bufferSize
        container.setMaxTextMessageBufferSize(1024000);
        container.setMaxBinaryMessageBufferSize(1024000);
        container.setMaxSessionIdleTimeout(15 * 60000L);
        return container;
    }

}


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

通信服务：

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

    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        SESSIONS.add(session);
        System.out.println("WebSocket消息】有新的连接，总数为：" + SESSIONS.size());
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        SESSIONS.remove(session);
        System.out.println("【WebSocket消息】连接断开，总数为：" + SESSIONS.size());
    }

    @SneakyThrows
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message){
        // Handle incoming text messages here, if needed
        String textMessage = message.getPayload();
//        System.out.println("Received text message: " + textMessage);

        // Broadcast the received text message to all connected clients
        for (WebSocketSession clientSession : SESSIONS) {
                if (clientSession.isOpen() && !clientSession.equals(session)) {
                    clientSession.sendMessage(new TextMessage(textMessage));
                }

        }
    }

    @Override
    protected 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();
            }
        }
    }
}


1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

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

  <img
              class="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 closed
    var reconnect_count=0
   
    reconnect_count += 1
    if (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) {
};
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48