接受websocket并定时检测连接状态及重连

whycxzp

2025-05-28 559f070ad00a82ccbff884c0192729f4fee74525

接受websocket并定时检测连接状态及重连

 src/main/java/com/whyc/webSocket/receiver/ThirdPartyWebSocketHandler.java

New file
@@ -0,0 +1,45 @@
package com.whyc.webSocket.receiver;

import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketHttpHeaders;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ThirdPartyWebSocketHandler extends TextWebSocketHandler {

    private static final Logger log = LoggerFactory.getLogger(ThirdPartyWebSocketHandler.class);
    private WebSocketSession session;

    @Override
    public void afterConnectionEstablished(WebSocketSession session) {
        this.session = session;
        log.info("WebSocket connection established.");
    }

    @Override
    public void handleTextMessage(WebSocketSession session, TextMessage message) {
        log.info("Received message: {}", message.getPayload());
        // 可以在这里做业务处理，比如发布事件、存入队列等
        String jsonStr = message.getPayload();

    }

    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) {
        log.error("Transport error occurred.", exception);
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
        log.info("WebSocket connection closed: {}", status);
        this.session = null;
    }

    public WebSocketSession getSession() {
        return session;
    }
}


 src/main/java/com/whyc/webSocket/receiver/WebSocketClientConfig.java

New file
@@ -0,0 +1,22 @@
package com.whyc.webSocket.receiver;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
import org.springframework.web.socket.WebSocketHttpHeaders;
import org.springframework.web.socket.sockjs.client.SockJsClient;
import org.springframework.web.socket.sockjs.client.Transport;
import org.springframework.web.socket.sockjs.client.WebSocketTransport;
import java.util.ArrayList;
import java.util.List;

@Configuration
public class WebSocketClientConfig {

    @Bean
    public ThirdPartyWebSocketHandler thirdPartyWebSocketHandler() {
        return new ThirdPartyWebSocketHandler();
    }

}


 src/main/java/com/whyc/webSocket/receiver/WebSocketClientService.java

New file
@@ -0,0 +1,56 @@
package com.whyc.webSocket.receiver;

import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
import org.springframework.web.socket.WebSocketHttpHeaders;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import java.net.URI;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Service
public class WebSocketClientService {

    /*@Value("${third.party.websocket.uri}")
    private String webSocketUri;*/

    @Autowired
    private ThirdPartyWebSocketHandler handler;

    public WebSocketClientService(ThirdPartyWebSocketHandler handler) {
        this.handler = handler;
    }

    private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

    @PostConstruct
    public void connect() {
        scheduler.scheduleAtFixedRate(() -> {
            try {
                //创建URI对象并将webSocketUri传入
                //String webSocketUri = "ws://localhost:8919/fg/server";
                String webSocketUri = "ws://localhost:18080/realtime-data";
                URI  uri = new URI(webSocketUri);
                System.out.println("webSocketUri:"+webSocketUri);
                if (handler.getSession() == null || !handler.getSession().isOpen()) {
                    System.out.println("Connecting to WebSocket server...");
                    StandardWebSocketClient client = new StandardWebSocketClient();
                    ListenableFuture<WebSocketSession> future = client.doHandshake(handler, new WebSocketHttpHeaders(), uri);
                    future.addCallback(session -> {
                        System.out.println("Connected to WebSocket server.");
                    }, ex -> {
                        System.err.println("Failed to connect to WebSocket: " + ex.getMessage());
                    });
                }
            } catch (Exception e) {
                System.err.println("Unexpected error during connection attempt: " + e.getMessage());
            }
        }, 0, 60, TimeUnit.SECONDS); // 每60秒尝试重连一次
    }
}