SpringBoot 集成 WebSocket(1)

简介: SpringBoot 集成 WebSocket

前言

       最近在做一个 WebSocket 通信服务的软件,所以必须跟着学一学。

1、WebSocket 概述

       一般情况下,我们的服务器和服务器之间可以发送请求,但是服务器是不能向浏览器去发送请求的。因为设计之初并没有想到以后会出现服务端频繁向客户端发送请求的情况。

       全双工的通信协议(WebSocket 最大的特点是浏览器也可以往服务器发请求,服务器也可以向浏览器发请求)。

1.1、浏览器和服务器使用WebSocket通信流程

1. 浏览器发起http请求,请求建立 WebSocket 连接

这里的协议升级就是说,我想通过这个 http 连接去升级为 WebSocket 连接

2. 服务器响应统一协议更改

3. 相互发送数据

升级了协议之后浏览器就可以和服务器相互通信了:

1.2、总结

  • WebSocket 协议是建立在 tcp 协议基础上的,所以不同语言也都支持
  • tcp 协议是全双工协议,http 协议基于它是单向的
  • WebSocket 没有同源限制,所以前后端端口不一致也不影响信息的发送

2、Java 实现 WebSocket 的两种方式

2.1、基于注解实现WebSocket服务器端

服务终端类

  • @ServerEndpoint:监听连接(需要传递一个地址参数)
  • @OnOpen:连接成功
  • @OnClose:连接关闭
  • @OnMessage:收到消息

配置类

  • 把 Spring 中的 ServerEndpointExporter 对象注入进来

2.2.1、编写服务终端类

// 监听哪些客户端来连接了WebSocket服务端
// 监听websocket地址 /myWs
@ServerEndpoint("/myWs")
@Component
@Slf4j
public class WebServerEndpoint {
 
    // 因为可能有多个客户端所以这里需要保证线程安全
    static Map<String,Session> sessionMap = new ConcurrentHashMap<>();
 
    // 建立连接时执行的操作
    @OnOpen
    public void onOpen(Session session){   // 每个websocket连接对于服务端来说都是一个Session
        sessionMap.put(session.getId(),session);
        log.info("websocket is open");
    }
 
    /**
     * 收到客户端消息时执行的操作
     * @param text 接受到客户端的消息
     * @return 返回给客户端的消息
     */
    @OnMessage
    public String onMessage(String text){
        log.info("收到一条新消息: " + text);
        return "收到 !";
    }
 
    // 连接关闭时执行的操作
    @OnClose
    public void onClose(Session session){
        sessionMap.remove(session.getId());
        log.info("websocket is close");
    }
 
    @Scheduled(fixedRate = 2000) // 每隔2s执行一次
    public static void sendMessage() throws IOException {
        for(String key: sessionMap.keySet()){ // 给所有客户端发送消息
            sessionMap.get(key).getBasicRemote().sendText("beat");
        }
    };
}

注意:这里监听的地址不可以是 "ws" 不然会报错,可能这是关键字吧,毕竟我们的协议就叫 ws 。

2.2.2、编写配置类

// 需要注入Bean的话必须声明为配置类
@Configuration
public class WebSocketConfig {
    @Bean
    public ServerEndpointExporter serverEndpointExporter(){
        return new ServerEndpointExporter();
    }
}

2.2、HTML + JS 实现客户端

在 resources 目录下创建 static/ws-client.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>WebSocket Client</title>
</head>
<script>
    // 客户端和服务器连接的地址(我们服务端监听的地址)
    let ws = new WebSocket("ws://localhost:8080/myWs1")
    ws.onopen=function (){ // 连接打开的时候向服务器发送一条消息
        ws.send("hey man")
    }
    ws.onmessage=function (message) {
        console.log(message.data)
    }
</script>
<body>
<h1>WebSocket</h1>
</body>
</html>

测试:启动 SpringBoot 并访问  localhost:8080/ws-client.html

执行结果:

2.3、基于 Spring 框架实现 WebSocket 服务器端

Spring 提供的类和框架

  • HttpSessionHandshakeInterceptor(抽象类):握手拦截器,在握手前后添加操作
  • AbstractWebSocketHandler(抽象类):WebSocket 处理程序,监听连接前、中、后
  • WebSocketConfigurer(接口):配置程序,比如配置监听哪个端口,配置自定义的握手拦截器,配置我们自定义的处理程序

2.3.1、编写握手拦截器类

/**
 * WebSocket 自定义握手拦截器
 */
@Component
@Slf4j
public class MyInterceptor extends HttpSessionHandshakeInterceptor {
    @Override
    public boolean beforeHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Map<String, Object> attributes) throws Exception {
        log.info("握手前");
        log.info("远程地址 => " + request.getRemoteAddress());
        // 保留父类的操作
        return super.beforeHandshake(request, response, wsHandler, attributes);
    }
 
    @Override
    public void afterHandshake(ServerHttpRequest request, ServerHttpResponse response, WebSocketHandler wsHandler, Exception ex) {
        log.info("完成握手");
        // 完成握手后它就会把 HTTP 协议升级为 WebSocket 协议
        super.afterHandshake(request, response, wsHandler, ex);
    }
}

2.3.2、编写 WebSocket 处理程序

/**
 * WebSocket 自定义处理程序
 */
@Slf4j
@Component
public class MyWsHandler extends AbstractWebSocketHandler {
    // WebSocketSession 对象可以封装一下吧用户的信息封装进去
    private static Map<String, SessionBean> sessionMap = new ConcurrentHashMap<>();
    // 线程安全的int值
    private static AtomicInteger clientIdMaker = new AtomicInteger(0);
 
    // 连接建立
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        super.afterConnectionEstablished(session);
        // 放在父方法调用之后
        SessionBean sessionBean = new SessionBean(session, clientIdMaker.getAndIncrement());
        sessionMap.put(session.getId(),sessionBean);
        log.info(sessionMap.get(session.getId()) + " connected");
    }
 
    // 收到消息
    @Override
    protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
        super.handleTextMessage(session, message);
        log.info(sessionMap.get(session.getId()).getClientId() + " : " + message.getPayload());
    }
 
    // 传输异常
    @Override
    public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
        super.handleTransportError(session, exception);
        // 如果异常就关闭 session
        if (session.isOpen())
            session.close();
        sessionMap.remove(session.getId());
    }
 
    // 连接关闭
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        super.afterConnectionClosed(session, status);
        log.info(sessionMap.get(session.getId()) + " closed");
        sessionMap.remove(session.getId());
    }
 
    @Scheduled(fixedRate = 2000) // 每隔2s执行一次
    public static void sendMessage() throws IOException {
        for(String key: sessionMap.keySet()){ // 给所有客户端发送消息
            sessionMap.get(key).getWebSocketSession().sendMessage(new TextMessage("beat"));
        }
    }
}

2.3.3、编写配置类

@Configuration
@EnableWebSocket    // 启用 spring 提供的 websocket 功能
public class MyWsConfig implements WebSocketConfigurer {
 
    @Resource
    MyWsHandler myWsHandler; // 引入我们在MyWsHandler上声明的Bean(@Component)
    @Resource
    MyInterceptor myInterceptor;
    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        // 监听的地址
        registry.addHandler(myWsHandler,"/myWs1")
                .addInterceptors(myInterceptor)
                .setAllowedOrigins("*"); // 允许的源
    }
}

修改 html 中的 websocket 地址为 /myWs1

测试:访问 localhost:8080/ws-client.html

3、总结

服务器会和每个客户端维护一个连接 :

3、WebSocket 实现多人聊天室

3.1、需求

  • 进入聊天室
  • 群聊功能,任何人说话,所有人都能接受到消息
  • 退出群聊

SpringBoot 集成 WebSocket(2)https://developer.aliyun.com/article/1534234

相关文章
|
9天前
|
前端开发 网络协议 JavaScript
在Spring Boot中实现基于WebSocket的实时通信
在Spring Boot中实现基于WebSocket的实时通信
|
9天前
|
数据可视化 Java API
Spring Boot与Swagger的集成
Spring Boot与Swagger的集成
|
9天前
|
Java API 开发者
在Spring Boot中集成Swagger API文档
在Spring Boot中集成Swagger API文档
|
1天前
|
Java 关系型数据库 MySQL
如何实现Springboot+camunda+mysql的集成
【7月更文挑战第2天】集成Spring Boot、Camunda和MySQL的简要步骤: 1. 初始化Spring Boot项目,添加Camunda和MySQL驱动依赖。 2. 配置`application.properties`,包括数据库URL、用户名和密码。 3. 设置Camunda引擎属性,指定数据源。 4. 引入流程定义文件(如`.bpmn`)。 5. 创建服务处理流程操作,创建控制器接收请求。 6. Camunda自动在数据库创建表结构。 7. 启动应用,测试流程启动,如通过服务和控制器开始流程实例。 示例代码包括服务类启动流程实例及控制器接口。实际集成需按业务需求调整。
|
1天前
|
SQL Java 数据库
实时计算 Flink版产品使用问题之Spring Boot集成Flink可以通过什么方式实现通过接口启动和关闭Flink程序
实时计算Flink版作为一种强大的流处理和批处理统一的计算框架,广泛应用于各种需要实时数据处理和分析的场景。实时计算Flink版通常结合SQL接口、DataStream API、以及与上下游数据源和存储系统的丰富连接器,提供了一套全面的解决方案,以应对各种实时计算需求。其低延迟、高吞吐、容错性强的特点,使其成为众多企业和组织实时数据处理首选的技术平台。以下是实时计算Flink版的一些典型使用合集。
|
8天前
|
Java API 开发工具
Spring Boot与Spring Cloud Config的集成
Spring Boot与Spring Cloud Config的集成
|
13天前
|
开发框架 Java 数据库
Spring Boot集成多数据源的最佳实践
Spring Boot集成多数据源的最佳实践
|
13天前
|
NoSQL Java MongoDB
Spring Boot与MongoDB的集成应用
Spring Boot与MongoDB的集成应用
|
14天前
|
消息中间件 Java 测试技术
【RocketMQ系列八】SpringBoot集成RocketMQ-实现普通消息和事务消息
【RocketMQ系列八】SpringBoot集成RocketMQ-实现普通消息和事务消息
25 1
|
8天前
|
消息中间件 Java Kafka
Spring Boot与Apache Kafka Streams的集成
Spring Boot与Apache Kafka Streams的集成