WebSocket-java篇

问题引入

消息推送的方式

我们要实现，服务器把消息推送到客户端，可以轮训，长轮训

还有sse

WebSocket理论

WebSocket 的由来与核心价值

• 诞生背景：解决 HTTP 协议在实时通信中的固有缺陷（单向请求-响应模式）

• 核心驱动力：

  • 替代低效轮询（Polling）和长轮询（Comet）

  • 满足实时应用需求（聊天、金融行情、游戏等）

• 核心优势：

  • 全双工通信：客户端/服务器可同时发送数据

  • 低延迟：从 HTTP 的数百 ms 降至 10-50ms

  • 高效传输：头部开销仅 2-14 字节（vs HTTP 的数百字节）

• 标准化：2011 年 RFC 6455 成为正式标准

WebSocket 协议核心组成

组成部分	作用	必要性
握手阶段	通过 HTTP 协议升级协商（101 Switching Protocols）切换到 WebSocket 协议	兼容现有网络基础设施（代理、防火墙）
数据帧	传输应用数据（文本/二进制）	封装数据，支持分片传输大文件
控制帧	管理连接状态（Ping/Pong 保活、Close 关闭）	维持连接健康，避免资源泄漏
掩码机制	客户端发送数据时进行 XOR 掩码加密	防止恶意代理缓存污染（安全关键）
Opcode	标识帧类型（文本/二进制/控制帧）	正确解析消息内容
Payload Length	动态长度标识（7/16/64位）	支持从短消息到 GB 级大文件传输

Spring Boot 深度集成方案

基础架构

核心组件详解

1. Client（客户端）

   • 作用：发起连接、订阅频道、收发消息

   • 为什么需要：作为通信的终端用户界面

   • 解决问题：

     • 提供用户交互入口

     • 实现跨平台通信（Web/App/桌面）

   • 技术实现：

     const socket = new WebSocket("ws://yourdomain/ws-endpoint");
     socket.onmessage = (event) => {console.log("收到消息:", event.data);
     };

2. Endpoint（连接端点）

   • 作用：处理握手请求，建立持久连接

   • 为什么需要：作为WebSocket连接的入口网关

   • 解决问题：

     • 协议升级（HTTP→WebSocket）

     • 连接生命周期管理

     • 跨域处理（CORS）

   • Spring Boot实现：

     @Configuration
     @EnableWebSocketMessageBroker
     public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {@Overridepublic void registerStompEndpoints(StompEndpointRegistry registry) {registry.addEndpoint("/ws-endpoint").setAllowedOrigins("*").withSockJS(); // 浏览器兼容方案}
     }

3. WebSocket Connection（连接管道）

   • 作用：维护全双工通信通道

   • 为什么需要：突破HTTP的无状态限制

   • 解决问题：

     • 避免频繁握手（单次握手持久连接）

     • 支持双向实时通信

     • 降低延迟（从HTTP的300ms+降至30ms内）

4. Message Broker（消息代理）

   • 作用：消息路由、分发、存储

   • 为什么需要：解耦生产者和消费者

   • 解决问题：

     • 海量连接下的消息分发

     • 分布式系统扩展

     • 消息持久化与重试

   • 配置示例：

     @Override
     public void configureMessageBroker(MessageBrokerRegistry registry) {// 使用外部消息中间件registry.enableStompBrokerRelay("/topic", "/queue").setRelayHost("rabbitmq-host").setRelayPort(61613);
     }

5. 频道系统（路由核心）

   频道类型	前缀	作用	解决的问题	消息流向
   广播频道	/topic	公共消息广播	1:N 消息分发 （如聊天室公告）	发布者 → 所有订阅者
   私有队列	/queue	点对点通信	1:1 精准投递 （如订单通知）	发布者 → 特定订阅者
   用户频道	/user	用户级隔离	多设备同步 （如微信网页/App同时在线）	发布者 → 用户所有会话

6. @MessageMapping Controller（业务处理器）

   • 作用：处理业务逻辑，生成响应

   • 为什么需要：分离通信协议与业务逻辑

   • 解决问题：

     • 业务逻辑集中管理

     • 消息验证与转换

     • 数据库/服务集成

   • 示例：

     @MessageMapping("/trade")
     @SendTo("/topic/stock-updates")
     public StockUpdate handleTrade(Order order) {// 1. 验证订单// 2. 执行交易// 3. 生成市场数据更新return tradingService.execute(order);
     }

架构演进价值

1. 协议层优化

   • 替代方案对比：

     方案	延迟	开销	双向通信	频道支持
     HTTP轮询	300ms+	高	❌	❌
     WebSocket基础	50ms	低	✔️	❌
     WS+STOMP	30ms	中	✔️	✔️

2. 工程化价值

3. 业务场景适配

   • 广播场景：/topic/news（新闻推送）

   • 私有场景：/queue/user-123/notifications（个人通知）

   • 混合场景：/topic/room-{id} + /user/queue/private（在线教育平台）

总结：为什么需要此架构

1. 连接管理 通过Endpoint统一处理握手/断开，解决连接生命周期管理混乱问题

2. 消息路由 频道系统实现发布-订阅模式，解决海量消息精准投递问题

3. 业务解耦 控制器隔离业务逻辑与通信协议，解决代码维护困难问题

4. 水平扩展 消息代理支持集群部署，解决单点性能瓶颈问题

5. 安全管控 频道级权限控制，解决敏感数据泄露风险

终极价值：此架构在协议层实现高效实时通信，在架构层通过频道机制解决复杂业务场景的消息路由问题，在工程层通过Spring Boot实现企业级标准化，是构建现代实时应用的基石。

原理流程

在我的E盘的WebSocket文件夹

消息执行流程（Flow）概览

建立连接（connect，连接）

Client（客户端）发起到 /ws-endpoint 的 WebSocket 握手（handshake，握手），Endpoint（端点）完成升级后建立 WebSocket Connection（WebSocket 连接）。

订阅频道（subscribe，订阅）

Client 通过 STOMP 向 broker 发送 SUBSCRIBE Frame（订阅帧），表示“我要订阅 /topic/greetings”。

发送消息到 Controller（SEND Frame）

Client 发送 SEND Frame（发送帧），destination（目的地）为 /app/hello。

Broker 根据 setApplicationDestinationPrefixes("/app")，将消息路由（route，路由）给匹配 @MessageMapping("/hello") 的方法

Controller（控制器）处理

GreetingController.handleHello(...) 被调用（invoke，调用），执行业务逻辑，返回 Greeting 对象。

Broker（代理）转发

因为方法上有 @SendTo("/topic/greetings")，返回值被封装成 MESSAGE Frame（消息帧）发送给 Broker（消息代理）。

Broker 将该消息分发（dispatch，分发）给所有订阅（subscription，订阅）了 /topic/greetings 的客户端 session。

Client（客户端）接收（receive，接收）

Client 在订阅回调（callback，回调）中拿到服务器推送（push，推送）的消息并渲染到页面。

这就是完整的一次流程。

API

客户端

websocket对象创建

let ws = new WebSocket(URL);

URL说明

格式：协议://ip地址:端口/访问路径 协议：协议名称为 ws

websocket对象相关事件

事件	事件处理程序	描述
open	ws.onopen	连接建立时触发
message	ws.onmessage	客户端接收到服务器发送的数据时触发
close	ws.onclose	连接关闭时触发

websocket对象提供的方法

方法名称	描述
send()	通过websocket对象调用该方法发送数据给服务端

简单示例

<script>
let ws = new WebSocket("ws://localhost/chat");
ws.onopen = function() {
};ws.onmessage = function(evt) {// 通过 evt.data 可以获取服务器发送的数据
};ws.onclose = function() {
};
</script>

服务端

Tomcat的7.0.5版本开始支持WebSocket,并且实现了Java WebSocket规范。

Java WebSocket应用由一系列的Endpoint组成。Endpoint是一个java对象，代表WebSocket链接的一端，对于服务端，我们可以视为处理具体WebSocket消息的接口。

我们可以通过两种方式定义Endpoint:

• 第一种是编程式，即继承类javax.websocket.Endpoint并实现其方法。

• 第二种是注解式，即定义一个POJO，并添加@ServerEndpoint相关注解。

Endpoint实例在WebSocket握手时创建，并在客户端与服务端链接过程中有效，最后在链接关闭时结束。在Endpoint接口中明确了与其生命周期相关的方法，规范实现者确保生命周期的各个阶段调用实例的相关方法。生命周期方法如下：

方法	描述	注解
onOpen()	当开启一个新的会话时调用，该方法是客户端与服务端握手成功后调用的方法	@OnOpen
onClose()	当会话关闭时调用	@OnClose
onError()	当连接过程异常时调用	@OnError

服务端如何接收客户端发送的数据呢？

• 编程式 通过添加 MessageHandler 消息处理器来接收消息

• 注解式 在定义 Endpoint 时，通过 @OnMessage 注解指定接收消息的方法

服务端如何推送数据给客户端呢？

发送消息则由 RemoteEndpoint 完成，其实例由 Session 维护。

发送消息有 2 种方式发送消息

• 通过 session.getBasicRemote 获取同步消息发送的实例，然后调用其 sendXxx() 方法发送消息

• 通过 session.getAsyncRemote 获取异步消息发送实例，然后调用其 sendXxx() 方法发送消息

@ServerEndpoint("/chat")
@Component
public class ChatEndpoint {@OnOpen// 连接建立时被调用public void onOpen(Session session, EndpointConfig config) {}@OnMessage// 接收到客户端发送的数据时被调用public void onMessage(String message) {}@OnClose// 连接关闭时被调用public void onClose(Session session) {}
}

WebSocket 消息分发的三种常见模式

session.getAsyncRemote()（getBasicRemote）.sendXxx() 方法本身并不直接区分这些模式，而是通过 目标地址（如 Session、Broadcast） 和 应用层逻辑 来实现不同的消息分发方式。

WebSocket 消息分发的三种常见模式

1. 单播（Unicast）

• 点对点发送：消息直接发送给某个特定的客户端（Session）。

• 实现方式：通过目标客户端的 session.getAsyncRemote().sendText()。

• 示例：

  // 向特定客户端发送消息
  targetSession.getAsyncRemote().sendText("Private message");

2. 广播（Broadcast）

• 一对多发送：消息发送给所有连接的客户端（或特定分组）。

• 实现方式：遍历所有 Session 或使用 @ServerEndpoint 的全局集合。

• 示例：

  // 广播给所有客户端
  for (Session session : allSessions) {session.getAsyncRemote().sendText("Broadcast message");
  }

• 注意：Java WebSocket API 本身不提供原生广播方法，需自行维护 Session 集合。

3. 组播（Multicast）

• 分组发送：消息发送给订阅了特定主题（Topic）或频道的客户端。

• 实现方式：通过应用层维护分组映射（如 Map<String, Set<Session>>）。

• 示例：

  // 向订阅了 "news" 频道的客户端发送消息
  for (Session session : channelSubscribers.get("news")) {session.getAsyncRemote().sendText("News update");
  }

总结

模式	目标范围	实现关键	适用场景
单播	单个 Session	直接调用目标 Session	私聊、定向通知
广播	所有 Session	遍历全局 Session 集合	公告、全局状态更新
组播	分组 Session	维护分组映射（Topic → Sessions）	频道订阅、房间聊天

WebSocket 的灵活性在于：sendXxx() 是工具，分发模式由开发者通过 Session 代码管理逻辑实现。

在线聊天室实现

具体代码在learnWebSocket里面

流程分析

package com.learnwebsocket.config;import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;/*** @version v1.0* @ClassName: WebsocketConfig*/
@Configuration
public class WebsocketConfig {/*** 创建一个ServerEndpointExporter对象，这个对象会自动注册使用了@ServerEndpoint注解的类* @return*/@Beanpublic ServerEndpointExporter serverEndpointExporter() {return new ServerEndpointExporter();}
}

后端

ServerEndpointExporter

首先，由于websocket不直接归于spring管理，属于spring的扩展模块，所以为了把websocket的实例也注册到spring里面，我们需要一个spring和websocket的连接桥梁。也就是ServerEndpointExporter。这个类负责加载websocket的端点。他同时可以被spring直接管理。

package com.learnwebsocket.config;import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.socket.server.standard.ServerEndpointExporter;/*** @version v1.0* @ClassName: WebsocketConfig*/
@Configuration
public class WebsocketConfig {/*** 创建一个ServerEndpointExporter对象，这个对象会自动注册使用了@ServerEndpoint注解的类* @return*/@Beanpublic ServerEndpointExporter serverEndpointExporter() {return new ServerEndpointExporter();}
}

端点Endpoint

然后。我们需要自己创建一个端点，供ServerEndpointExporter发现管理。

这里面我们需要实现三个方法，这个上面有讲。

这里面还有广播和单播的实现代码，仔细看看。

还有的就是，由于Endpoint不直接属于spring，若要给Endpoint去配置一些东西，我们需要手动创建一个类，实现java给我们的接口，来去配置之后给spring管理

package com.learnwebsocket.ws.pojo;import com.alibaba.fastjson.JSON;import com.learnwebsocket.config.GetHttpSessionConfig;
import com.learnwebsocket.utils.MessageUtils;
import org.springframework.stereotype.Component;import javax.servlet.http.HttpSession;
import javax.websocket.*;
import javax.websocket.server.ServerEndpoint;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;/*** @version v1.0* @ClassName: ChatEndpoint* @Description: 端点* @Author: 黑马程序员*/
@ServerEndpoint(value = "/chat",configurator = GetHttpSessionConfig.class)
@Component
public class ChatEndpoint {// 用来保存所有的用户private static final Map<String,Session> onlineUsers = new ConcurrentHashMap<>();//当前用户对应的session对象private HttpSession httpSession;/*** 建立websocket连接后，被调用* @param session*/@OnOpenpublic void onOpen(Session session, EndpointConfig config) {//1，将session进行保存this.httpSession = (HttpSession) config.getUserProperties().get(HttpSession.class.getName());String user = (String) this.httpSession.getAttribute("user");onlineUsers.put(user,session);//2，广播消息。需要将登陆的所有的用户推送给所有的用户String message = MessageUtils.getMessage(true,null,getFriends());broadcastAllUsers(message);}public Set getFriends() {Set<String> set = onlineUsers.keySet();return set;}// 广播所有用户private void broadcastAllUsers(String message) {try {//遍历map集合Set<Map.Entry<String, Session>> entries = onlineUsers.entrySet();for (Map.Entry<String, Session> entry : entries) {//获取到所有用户对应的session对象Session session = entry.getValue();//发送消息session.getBasicRemote().sendText(message);}} catch (Exception e) {//记录日志}}/*** 浏览器发送消息到服务端，该方法被调用** 张三  -->  李四* @param message*/@OnMessagepublic void onMessage(String message) {try {//将消息推送给指定的用户Message msg = JSON.parseObject(message, Message.class);//获取 消息接收方的用户名String toName = msg.getToName();String mess = msg.getMessage();//获取消息接收方用户对象的session对象Session session = onlineUsers.get(toName);String user = (String) this.httpSession.getAttribute("user");String msg1 = MessageUtils.getMessage(false, user, mess);session.getBasicRemote().sendText(msg1);} catch (Exception e) {//记录日志}}/*** 断开 websocket 连接时被调用* @param session*/@OnClosepublic void onClose(Session session) {//1,从onlineUsers中剔除当前用户的session对象String user = (String) this.httpSession.getAttribute("user");onlineUsers.remove(user);//2,通知其他所有的用户，当前用户下线了String message = MessageUtils.getMessage(true,null,getFriends());broadcastAllUsers(message);}
}

配置类

上面的httpSession来自配置类的，因为登陆后我们把用户的名字存到了httpSession。但是websocket无法直接获取httpSession，所以要把它存到websocket配置文件里面。再获取。

package com.learnwebsocket.config;import javax.servlet.http.HttpSession;
import javax.websocket.HandshakeResponse;
import javax.websocket.server.HandshakeRequest;
import javax.websocket.server.ServerEndpointConfig;/*** @version v1.0* @ClassName: GetHttpSessionConfig*/
public class GetHttpSessionConfig extends ServerEndpointConfig.Configurator {@Overridepublic void modifyHandshake(ServerEndpointConfig sec, HandshakeRequest request,HandshakeResponse response) {//获取HttpSession对象HttpSession httpSession = (HttpSession) request.getHttpSession();//将httpSession对象保存起来sec.getUserProperties().put(HttpSession.class.getName(),httpSession);}
}

前端

先登陆之后，然后向后端的端点请求websocket的连接，之后绑定三个方法。

await axios.get("user/getUsername").then(res => {this.username = res.data;});//创建webSocket对象ws = new WebSocket("ws://localhost:8080/chat");//给ws绑定事件ws.onopen = this.onopen;//接收到服务端推送的消息后触发ws.onmessage = this.onMessage;ws.onclose = this.onClose;