SpringBoot项目中使用WebSocket实现实时通信功能_Java

一、前言

在传统的 http 通信中，客户端发起请求，服务器给出响应，一次通信就此结束。这种模式对于静态页面展示完全够用，但对于需要实时推送的场景——如在线聊天、实时通知、股票行情、设备状态监控——就力不从心了。

早期的解决方案是轮询：客户端每隔几秒就发一次请求问服务器"有新消息吗"，效率低下且浪费资源。websocket 的出现彻底改变了这一局面。

二、websocket 是什么

websocket 是一种在单个 tcp 连接上进行全双工通信的网络协议，由 html5 规范引入，rfc 6455 正式定义。

2.1 与 http 的核心区别

对比项	http	websocket
通信方向	单向（客户端请求，服务器响应）	双向（任意一方均可主动发送）
连接状态	无状态，一问一答后断开	有状态，连接建立后持续保持
协议头开销	每次请求都携带完整 header	握手一次后，后续帧头极小（2~10字节）
适用场景	普通页面请求	实时推送、聊天、游戏

2.2 握手过程

websocket 复用了 http 的握手机制，通过一次 http 请求完成协议升级：

客户端发送升级请求：
get /ws/chat http/1.1
host: localhost:8080
upgrade: websocket
connection: upgrade
sec-websocket-key: dghlihnhbxbszsbub25jzq==
sec-websocket-version: 13
服务器返回：
http/1.1 101 switching protocols
upgrade: websocket
connection: upgrade
sec-websocket-accept: s3pplmbitxaq9kygzzhzrbk+xoo=

握手成功后，协议从 http 升级为 websocket，后续通信不再经过 http 层，直接在 tcp 连接上传输 websocket 帧。

三、spring boot 集成 websocket

spring boot 提供了两种方式使用 websocket：

方式一：基于 java ee 标准的 @serverendpoint（本文重点介绍）
方式二：基于 spring 的 websockethandler

3.1 添加依赖

<dependency>
    <groupid>org.springframework.boot</groupid>
    <artifactid>spring-boot-starter-websocket</artifactid>
</dependency>

3.2 注册配置类

使用内置 tomcat 时，需要手动注册 serverendpointexporter，让 spring 能扫描到 @serverendpoint 注解：

@configuration
public class websocketconfig {
    /**
     * 使用内置 tomcat 时必须注入此 bean
     * 若使用外部 tomcat 部署，则不需要，容器会自行管理
     */
    @bean
    public serverendpointexporter serverendpointexporter() {
        return new serverendpointexporter();
    }
}

3.3 编写 websocket 端点

@component
@serverendpoint("/ws/chat/{userid}")
public class chatwebsocket {
    private static final logger log = loggerfactory.getlogger(chatwebsocket.class);
    // 存储所有在线连接，key=userid，value=websocket会话
    private static final concurrenthashmap<string, session> sessions = new concurrenthashmap<>();
    /**
     * 连接建立时触发
     */
    @onopen
    public void onopen(session session, @pathparam("userid") string userid) {
        sessions.put(userid, session);
        log.info("用户 {} 上线，当前在线人数：{}", userid, sessions.size());
    }
    /**
     * 收到客户端消息时触发
     */
    @onmessage
    public void onmessage(string message, @pathparam("userid") string userid) {
        log.info("收到用户 {} 的消息：{}", userid, message);
        // 解析消息，转发给目标用户
        handlemessage(userid, message);
    }
    /**
     * 连接关闭时触发
     */
    @onclose
    public void onclose(@pathparam("userid") string userid) {
        sessions.remove(userid);
        log.info("用户 {} 下线，当前在线人数：{}", userid, sessions.size());
    }
    /**
     * 发生异常时触发
     */
    @onerror
    public void onerror(session session, throwable error) {
        log.error("websocket 发生异常：{}", error.getmessage());
    }
    /**
     * 向指定用户发送消息
     */
    public static void sendtouser(string userid, string message) {
        session session = sessions.get(userid);
        if (session != null && session.isopen()) {
            try {
                session.getbasicremote().sendtext(message);
            } catch (ioexception e) {
                log.error("向用户 {} 发送消息失败：{}", userid, e.getmessage());
            }
        }
    }
    /**
     * 广播消息给所有在线用户
     */
    public static void broadcast(string message) {
        sessions.foreach((userid, session) -> {
            if (session.isopen()) {
                try {
                    session.getbasicremote().sendtext(message);
                } catch (ioexception e) {
                    log.error("广播消息失败，userid={}：{}", userid, e.getmessage());
                }
            }
        });
    }
    /**
     * 处理消息逻辑（示例：解析json转发）
     */
    private void handlemessage(string fromuserid, string message) {
        try {
            // 假设消息格式为 json：{"touserid":"xxx","content":"hello"}
            jsonobject json = jsonobject.parseobject(message);
            string touserid = json.getstring("touserid");
            string content = json.getstring("content");
            jsonobject response = new jsonobject();
            response.put("fromuserid", fromuserid);
            response.put("content", content);
            response.put("time", system.currenttimemillis());
            sendtouser(touserid, response.tojsonstring());
        } catch (exception e) {
            log.error("消息处理失败：{}", e.getmessage());
        }
    }
}

3.4 在业务代码中主动推送

websocket 不只能被动接收消息，也可以在任意业务逻辑中主动向客户端推送：

@service
public class orderservice {
    public void completeorder(long orderid, long userid) {
        // 处理订单完成逻辑...
        // 订单完成后，主动推送通知给用户
        jsonobject notify = new jsonobject();
        notify.put("type", "order_complete");
        notify.put("orderid", orderid);
        notify.put("message", "您的订单已完成，请及时查看");
        chatwebsocket.sendtouser(string.valueof(userid), notify.tojsonstring());
    }
}

四、前端对接

<!doctype html>
<html>
<body>
<script>
    const userid = "user_001";
    const ws = new websocket(`ws://localhost:8080/ws/chat/${userid}`);
    // 连接建立
    ws.onopen = function () {
        console.log("websocket 连接成功");
        ws.send(json.stringify({
            touserid: "user_002",
            content: "你好！"
        }));
    };
    // 收到消息
    ws.onmessage = function (event) {
        const data = json.parse(event.data);
        console.log(`收到来自 ${data.fromuserid} 的消息：${data.content}`);
    };
    // 连接关闭
    ws.onclose = function () {
        console.log("websocket 连接已关闭");
    };
    // 发生错误
    ws.onerror = function (error) {
        console.error("websocket 错误：", error);
    };
</script>
</body>
</html>

五、生产环境注意事项

5.1 @serverendpoint 无法注入 spring bean 的问题

@serverendpoint 的实例由 tomcat 管理，不是 spring bean，所以直接用 @autowired 注入会失败：

@serverendpoint("/ws/chat/{userid}")
public class chatwebsocket {
    @autowired
    private userservice userservice;  // ❌ 注入失败，值为 null
}

解决方案：通过静态变量 + applicationcontext 获取

@serverendpoint("/ws/chat/{userid}")
public class chatwebsocket implements applicationcontextaware {
    private static userservice userservice;
    @override
    public void setapplicationcontext(applicationcontext context) {
        userservice = context.getbean(userservice.class);
    }
}

或者更简洁地，在配置类里提前拿到：

@component
public class websocketbeanfactory implements applicationcontextaware {
    private static applicationcontext context;
    @override
    public void setapplicationcontext(applicationcontext applicationcontext) {
        context = applicationcontext;
    }
    public static <t> t getbean(class<t> clazz) {
        return context.getbean(clazz);
    }
}
// 在 websocket 端点里使用
userservice userservice = websocketbeanfactory.getbean(userservice.class);

5.2 集群部署下的消息广播问题

单机部署时，所有连接都在同一个 jvm 里，concurrenthashmap 存储会话没问题。但集群部署时，用户 a 连接在节点1，用户 b 连接在节点2，节点1无法直接找到用户 b 的 session。

解决方案：引入消息中间件（如 rabbitmq / redis pub-sub）

节点1收到消息
    ↓
发布到 mq / redis channel
    ↓
所有节点订阅并消费消息
    ↓
各节点检查自己管理的 session 里有没有目标用户
    ↓
有则发送，无则忽略

5.3 连接心跳保活

长时间无数据交换时，防火墙或负载均衡器可能会断开连接，需要定期发送心跳：

// 前端每 30 秒发送一次心跳
setinterval(() => {
    if (ws.readystate === websocket.open) {
        ws.send(json.stringify({ type: "ping" }));
    }
}, 30000);

// 后端收到心跳回复
@onmessage
public void onmessage(string message, session session) {
    jsonobject json = jsonobject.parseobject(message);
    if ("ping".equals(json.getstring("type"))) {
        try {
            session.getbasicremote().sendtext("{\"type\":\"pong\"}");
        } catch (ioexception e) {
            log.error("心跳回复失败");
        }
    }
}

5.4 连接断开重连

网络抖动时前端需要自动重连：

function createwebsocket(userid) {
    const ws = new websocket(`ws://localhost:8080/ws/chat/${userid}`);
    ws.onclose = function () {
        console.log("连接断开，3秒后重连...");
        settimeout(() => createwebsocket(userid), 3000);
    };
    return ws;
}

六、完整流程总结

1. 引入 spring-boot-starter-websocket 依赖
         ↓
2. 注册 serverendpointexporter bean（内置tomcat必须）
         ↓
3. 编写 @serverendpoint 端点类
   实现 @onopen / @onmessage / @onclose / @onerror
         ↓
4. 用 concurrenthashmap 管理所有在线 session
         ↓
5. 业务代码调用静态方法主动推送消息
         ↓
6. 前端用 new websocket(url) 建立连接
   通过 onmessage 接收，ws.send() 发送

七、适用场景总结

场景	说明
在线聊天	用户之间实时发送消息
实时通知	订单完成、充值到账、审批结果
设备监控	硬件设备实时上报状态数据
协同编辑	多人同时编辑同一文档
行情推送	股票、加密货币实时价格
在线游戏	多人游戏实时同步状态