SpringBoot集成WebSocket

SpringBoot集成WebSocket

1.引jar包

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-websocket</artifactId>
        </dependency>
        <dependency>

2.书写配置类

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;

/**
 * 配置信息
 */
@Configuration
public class WebsocketConfig {

    /**
     * 注入一个ServerEndpointExporter,该Bean会自动注册使用@ServerEndpoint注解声明的websocket endpoint
     * @date 2023/6/20 9:19
     * @return ServerEndpointExporter
     */
    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }

}

3.使用注解开发一个WebSocket接口类

import org.springframework.stereotype.Component;

import javax.websocket.OnClose;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 监听Websocket接口myWs
 */
@ServerEndpoint("/myWs")
@Component
public class WsServerEndpoint {

    private static Map<String,Session> sessions = new ConcurrentHashMap<>();
    /**
     * 连接成功
     *
     * @param session
     */
    @OnOpen
    public void onOpen(Session session) {
        System.out.println("打开连接");
        sessions.put(session.getId(),session);
    }

    /**
     * 连接关闭
     *
     * @param session
     */
    @OnClose
    public void onClose(Session session) {
        System.out.println("关闭连接");
        sessions.remove(session.getId());
    }

    /**
     * 接收到消息
     *
     * @param text
     */
    @OnMessage
    public void onMsg(String text) throws IOException {
        System.out.println("收到消息："+text);
        sendMessage();
    }

    /**
     * 发送信息
     */
    public void sendMessage(){
        sessions.forEach((k,v)->{
            try {
                v.getBasicRemote().sendText("hello");
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        });
    }
}

4.测试

​ 使用apipost或其他测试工具新建一个webSocket接口

​ webSocket接口地址：ws://localhost:8080/myWs

SpringBoot实现WebSocket接口转发

1.书写一个代理处理器

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.concurrent.ListenableFuture;
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.client.WebSocketClient;
import org.springframework.web.socket.client.standard.StandardWebSocketClient;
import org.springframework.web.socket.handler.AbstractWebSocketHandler;

/**
 * @author wangfan
 */
public class ProxyWebSocketHandler extends AbstractWebSocketHandler {

    private static  final Logger log = LoggerFactory.getLogger(ProxyWebSocketHandler.class);

    private final WebSocketClient client = new StandardWebSocketClient();

    private WebSocketSession targetSession;

    private WebSocketSession thisSession;

    private String targetUrl;


    public ProxyWebSocketHandler(String targetUrl){
        this.targetUrl = targetUrl;
        initTargetSession();
    }
    private void initTargetSession (){
        try {
            ListenableFuture<WebSocketSession> future = client.doHandshake(new ForwardingWebSocketHandler(), this.targetUrl);
            future.addCallback(new WebSocketConnectionCallback());
            targetSession = future.get();
        } catch (Exception e) {
            log.error("目标ws服务链接失败 " + e.getMessage());
        }
    }

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {

        log.info("链接本地会话成功");
        if(targetSession == null || (targetSession != null && !targetSession.isOpen())){
            initTargetSession();
        }else{
            log.info("链接目标会话成功");
        }
        //提取查询参数
        this.thisSession = session;

    }

    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        if (targetSession != null && targetSession.isOpen()) {
            log.info("转发文本消息 {}", message.getPayload());
            targetSession.sendMessage(message);
        } else {
            log.warn("T目标会话不存在或者关闭");
        }
    }

    @Override
    protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) throws Exception {
        if (targetSession != null && targetSession.isOpen()) {
            log.info("转发二进制消息");
            targetSession.sendMessage(message);
        } else {
            log.warn("目标会话不存在或者关闭");
        }
    }

    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        log.error("转换错误: {}", exception.getMessage());
        if (session.isOpen()) {
            session.close();
        }
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        log.info("链接关闭: {}", status);
        if (session != null && session.isOpen()) {
            session.close();
        }
    }

    private class ForwardingWebSocketHandler extends AbstractWebSocketHandler {
        @Override
        protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
            log.info("接收到的消息 {}", message.getPayload());
            thisSession.sendMessage(message);
            //ProxyWebSocketHandler.this.handleTextMessage(session, message);
        }

        @Override
        protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) throws Exception {
            ProxyWebSocketHandler.this.handleBinaryMessage(session, message);
        }
    }

    private class WebSocketConnectionCallback implements org.springframework.util.concurrent.ListenableFutureCallback<WebSocketSession> {
        @Override
        public void onSuccess(WebSocketSession result) {
            log.info("发布目标对象成功");
            targetSession = result;
        }

        @Override
        public void onFailure(Throwable t) {
            log.error("链接目标对象失败: {}", t.getMessage());
        }
    }
}

2，代理处理器放入容器并开启Websocket接口

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;

/**
 * @author wangfan
 */
@Configuration
@EnableWebSocket
public class ProxyWebSocketConfig implements WebSocketConfigurer {


    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(new ProxyWebSocketHandler("ws://localhost:8080/myWs"),"/webSocket").setAllowedOrigins("*");
    }
}

4.测试

​ 使用apipost或其他测试工具新建一个webSocket接口

​ webSocket接口地址：ws://localhost:8081/webSocket

​ 查看消息被转发到ws://localhost:8080/myWs，且收到该服务的消息