做后端开发的人基本都会碰到一类需求给某个指定用户推送一条消息或者朝全平台在线用户喊一嗓子。前者是私发后者是广播在Java生态里这俩经常被放到一起做组合成一个消息推送模块。私发类比微信单聊广播类比直播间的全量公告看起来都是“把一段文本从服务器发到浏览器/客户端”但真要落地成代码连接管理、消息协议、线程安全、断线重连任何一环偷懒都会在线上给你颜色看。这篇文章不绕弯子直接梳理“Java消息私发和广播”从设计到实现的完整链路。我会把核心会话管理、点对点消息、全员广播的实现思路掰开揉碎附带可以直接抄走的Spring Boot WebSocket Demo最后聊聊我实际踩过的坑。适合刚接触WebSocket的Java后端也适合正在把手写Socket升级成稳定模块的同学。1. 这个需求本质在解决什么问题1.1 私发和广播的场景拆解私发和广播虽然都叫“发消息”但背后是两套完全不同的消息分发模型。私发要求精确投递A用户发给B用户服务器必须先知道B在哪然后单独走一条通路把消息塞给B。这里的关键是“寻址”就像寄快递必须有收件人地址不能扔到小区门口就完事。广播要求的是扇出一条消息要复制N份推给所有当前在线的人。这里的关键是“遍历与容错”只要有一个连接出问题不能把整批发送都卡死。真实业务里这俩往往会混合出现。比如一个客服系统用户发给客服的消息是私发系统维护公告全员推送是广播某个工单状态变更既要通知提交人又要通知处理人这就是一组私发。想清楚使用场景再去设计代码才不会把接口写拧巴。1.2 技术选型为什么我选了WebSocket而不是轮询聊Java消息推送不止一种路子。最简单的就是HTTP轮询前端每隔几秒发一次请求问“有没有新消息”。这种方式优点是实现门槛极低缺点是服务器压力大、消息实时性差高峰期一群人狂点会直接把接口打崩体验也很一般。还有一个方案是SSEServer-Sent Events服务器单向往浏览器推适合公告流、通知流但浏览器端API和长连接管理做起来总觉得不太顺手。我最终选择WebSocket核心就三点全双工通信、一次握手长期复用、消息实时性拉满。WebSocket的优势不仅体现在“能发消息”更关键的是它能同时承载上行和下行。用户在线状态、心跳保活、客户端主动上报这些后续要加的功能都能在同一条连接上扩展。你当然可以用纯Java写一套原生的Socket 多线程来管理连接但Spring Boot自带的WebSocket Starter已经把端点注册、握手握手、连接生命周期这些底层逻辑封装好了我们只需要关注业务本身。1.3 一开始就定好消息模型后面能少改很多代码新手写消息推送最容易犯的错是拿到需求直接写一个sendText(String message)把消息原样丢出去结果没过几天就发现不对劲没人知道这条消息到底是要私发还是广播也没人知道消息体里该放什么字段。我在动手写代码前一定先把消息协议定好。哪怕只是Demo也要考虑消息类型、目标用户、消息内容、发送时间这几个基础字段。前端收到消息后先看消息类型再做对应处理后端解析时也能快速判断该走私发逻辑还是广播逻辑。有了这层约定后面接入前端、测试、线上排错都会省力得多。2. 核心会话管理设计与原理解析2.1 用一张线程安全的在线表管理所有连接消息推送要解决的头号问题是“服务器怎么知道用户当前连在哪”。HTTP是无状态的一个请求处理完连接就断WebSocket不一样连接会一直留着。于是我们必须自己维护一张“在线连接表”把用户标识和对应的WebSocket会话对应起来。这张表在设计上有几个硬性要求第一必须线程安全。WebSocket服务端是并发模型每来一条连接就会有一个线程或异步任务在处理多个用户同时上线、下线、收发消息肯定会有并发读写这张表。用普通的HashMap扩容和数据覆盖都是雷我习惯用ConcurrentHashMap读多写少的场景下它表现很稳。第二key要稳定唯一。选择用户ID作为key最直接但要注意区分业务场景。比如同一用户开两个浏览器标签页就有两个WebSocket连接到底以哪个为准这要在业务层面定策略要么同一个用户只保留最新连接旧连接主动关闭避免一个用户收到重复消息要么支持一个key对应多个session发送时全session都发。大多数后台管理系统选前者简单不易乱。第三value不能只存session对象。session里带了很多底层信息但光有个session还不够我们通常还会维护一个lastHeartbeat字段用来做心跳检测和踢掉死连接。所以我更倾向于用一个简单封装对象作为value而不是直接存裸session。伪代码大概是这样的Component public class SessionManager { private final ConcurrentHashMapString, ClientSession onlineUsers new ConcurrentHashMap(); public void add(String userId, Session session) { ClientSession clientSession new ClientSession(userId, session, System.currentTimeMillis()); onlineUsers.put(userId, clientSession); } public ClientSession remove(String userId) { return onlineUsers.remove(userId); } public Session getSession(String userId) { ClientSession clientSession onlineUsers.get(userId); return clientSession null ? null : clientSession.getSession(); } public CollectionClientSession getAllSessions() { return onlineUsers.values(); } public boolean isOnline(String userId) { return onlineUsers.containsKey(userId); } }这段代码看着简单但它是整个私发和广播模块的地基。后面所有发送逻辑都要围绕这张表展开。2.2 私发为什么必须查表定位私发的本质就是根据接收方ID到在线连接表里找对应的WebSocket会话然后推送消息。public boolean sendToUser(String userId, String message) { Session session sessionManager.getSession(userId); if (session null || !session.isOpen()) { return false; } session.getBasicRemote().sendText(message); return true; }这段代码里有两个容易被忽略的细节。第一个是session.isOpen()的判断。客户端可能已经断网断电服务端还没感知到如果把一条消息写到一条已经关闭的连接上会抛异常。所以在发送前必须先确认会话仍然有效。第二个是发送方法的返回。实时消息系统里“消息没发出去”是常态接收方离线、连接抖动、服务端重启都会导致投递失败。私发必须给出成功或失败的结果方便上层决定是落离线消息还是给发送方一个“对方不在线”的提示。如果直接把sendText包在try-catch里然后忽略结果短期看不出问题长期一定会积累一堆“消息发了但用户没收到”的线上反馈。2.3 广播为什么不能直接遍历所有连接广播看起来更简单把在线连接表遍历一遍挨个调用sendText就完事。但真正的广播需要考虑三个问题。问题一连接断开的隔离。遍历过程中如果某个sessionsendText抛出异常我见过不少人直接catch住就继续这是可以的但有个隐藏前提——你必须确认这个异常的session只是被跳过而不会影响后续其他session的发送。如果错误处理不好一个脏连接就把整个广播批次拖垮。所以我建议每个session独立try-catch这个原则和批量任务处理里的隔离思路是一样的。问题二消息风暴。如果在线用户有一万广播一次就要调用一万次sendText每次都是一次网络IO和底层协议封包。业务上线之后如果广播频率高比如每分钟广播一次服务器CPU和带宽都会面临压力。这种场景下广播不只是“发出去就完事”还需要考虑是否要限制频率、是否要对大文本做压缩这都属于生产环境必须补的课。问题三并发遍历与实时上下线冲突。遍历getAllSessions()时正好有人掉线可能迭代到已关闭的session并触发异常有人上线也可能出现拷贝迭代时拿到的是旧快照。ConcurrentHashMap在设计上弱一致迭代时不会抛并发修改异常但你要明白这并不代表迭代结果绝对实时。一个稳妥的广播方法长这样public void broadcast(String message) { for (ClientSession clientSession : sessionManager.getAllSessions()) { Session session clientSession.getSession(); if (session null || !session.isOpen()) { continue; } try { session.getBasicRemote().sendText(message); } catch (Exception e) { // 记录日志但不要中断整个广播 } } }这里的关键是先检查连接状态再单独发发错了只记录日志不抛出这样一条废连接不会拖累其他人。2.4 并发染上线程安全踩的坑HashMap转的教训这个坑我是真实踩过的。早期写消息推送的时候图省事直接用了HashMapString, Session上线那天几十个人同时连上来服务器console刷了一屏ConcurrentModificationException。问题就出在多个WebSocket线程同时put、同时遍历、同时removeHashMap在扩容时遇到并发写内部链表直接乱掉。解决方式很简单要么用ConcurrentHashMap要么用Collections.synchronizedMap包裹再加手动同步。前者代码最干净也是我最常用的一种。不过ConcurrentHashMap也不是万能比如“先判断再放入”这种复合操作仍然存在竞态条件// 有并发问题因为containsKey和put不是原子的 if (!onlineUsers.containsKey(userId)) { onlineUsers.put(userId, session); }真正需要原子注册的时候可以用putIfAbsent方法ClientSession old onlineUsers.putIfAbsent(userId, clientSession); if (old ! null) { // 说明用户已存在做去重或顶号处理 }同时连接、同时心跳、同时掉线这种多线程场景在消息系统里是常态所以在线连接表里每一步操作都需要考虑并发语义。3. 实操过程用Java WebSocket跑通一个私发和广播的Demo3.1 工程结构与依赖准备我直接用Spring Boot来搭这个Demo版本用Spring Boot 2.7.x或者3.x都可以核心依赖只需要两个spring-boot-starter-web和spring-boot-starter-websocket。前者提供HTTP接口方便我们做一个触发消息的入口后者提供WebSocket端点支持。这一步有一个非常重要的配置如果你用的是Spring Boot内置容器需要注册ServerEndpointExporter这个bean它负责扫描所有标注了ServerEndpoint注解的类。有些同学照着教程写了端点却始终访问不到八成就是漏了这个配置。pom里加上依赖后最重要的配置类如下Configuration public class WebSocketConfig { Bean public ServerEndpointExporter serverEndpointExporter() { return new ServerEndpointExporter(); } }这个配置就一行但是少写一行整个WebSocket都起不来排查的时候很容易被忽略。3.2 定义消息结构私发和广播先做区分消息结构我在前面强调过这一步真的不能省。这里我定义一个简单的消息封装类用Jackson序列化和反序列化public class ChatMessage { private String type; // private: 私发, broadcast: 广播 private String fromUserId; private String toUserId; private String content; private long timestamp; // 省略getter/setter }type字段是整个协议的开关。收到一条消息后先解析JSON拿到type再判断是走私发逻辑还是广播逻辑。这个设计的可扩展性也很好后面想加“群组消息”可以在type里扩展比如group再补充一个groupId字段代码结构不用推倒重来。另外我会习惯给消息加一个timestamp虽然Demo阶段看起来没用但真正出问题的时候对日志、排查延迟、确认顺序都靠这个字段。3.3 WebSocket处理器核心的消息入口端点类是整个模块的心脏所有连接建立、消息到达、连接关闭都会触发这里的方法。Component ServerEndpoint(/ws) public class MessageEndpoint { private Session session; private String userId; OnOpen public void onOpen(Session session) { this.session session; this.userId session.getRequestParameterMap().getOrDefault(userId, unknown).get(0); SessionManager.getInstance().add(userId, session); System.out.println(用户 userId 上线当前在线 SessionManager.getInstance().getOnlineCount()); } OnMessage public void onMessage(String message, Session session) { ChatMessage msg JSON.parseObject(message, ChatMessage.class); if (private.equals(msg.getType())) { // 私发 sendPrivate(msg); } else if (broadcast.equals(msg.getType())) { // 广播 broadcastMsg(msg); } } OnClose public void onClose(Session session) { SessionManager.getInstance().remove(userId); } OnError public void onError(Session session, Throwable error) { SessionManager.getInstance().remove(userId); } }这里有几个设计点需要重点解释。第一类是userId从哪来。Demo里我直接放在连接URL上客户端连接时带上?userId1001这种参数。实际项目中一定要换成token鉴权让后端从令牌中解析出用户身份而不是让客户端自己上报一个ID否则任何一个人都可以伪造别人ID来收消息这是Chat应用的重大安全漏洞。第二是onClose和onError里都要清理会话。很多同学只写了OnClose结果网络异常断开时连接表里残留了一堆僵尸session发消息的时候对着一个已经死掉的session一直报错。记住onError的触发频率远比你想象的高。第三是消息入口要try-catch。虽然框架层面不允许异常外抛但如果你在onMessage期间抛出了未捕获异常WebSocket连接会被直接关闭。所以解析消息、发送消息都建议包一层异常处理只记录日志不要让底层的IO异常把整个端点拖挂。3.4 私发与广播的发送实现接入层收到消息后真正的动作落在发送逻辑上。我把这俩方法单独抽出来方便复用private void sendPrivate(ChatMessage msg) { Session targetSession SessionManager.getInstance().getSession(msg.getToUserId()); if (targetSession null || !targetSession.isOpen()) { System.out.println(用户 msg.getToUserId() 不在线); return; } try { targetSession.getBasicRemote().sendText(JSON.toJSONString(msg)); } catch (IOException e) { System.err.println(私发消息失败 e.getMessage()); } } private void broadcastMsg(ChatMessage msg) { String messageText JSON.toJSONString(msg); for (ClientSession clientSession : SessionManager.getInstance().getAllSessions()) { if (clientSession.getSession().isOpen()) { try { clientSession.getSession().getBasicRemote().sendText(messageText); } catch (IOException e) { System.err.println(广播发送失败用户 clientSession.getUserId()); } } } }如果你把代码跑起来会发现一个有意思的问题私发的时候如果接收方不在线这段代码只是打一行日志就什么都不做了。真实业务里这显然不够所以生产环境中这里会触发离线消息存储把消息存进数据库或者消息队列等接收方上线后再拉取。私发和离线消息是一对孪生兄弟别只看在线发送这一层。3.5 加一个HTTP触发入口验证起来就方便了WebSocket是客户端主动连接后收消息但开发联调时总不可能每次都在另一个客户端里输入JSON去触发广播。我习惯在Demo里顺带提供两个HTTP接口用来模拟“管理端发私发”和“服务端发广播”。RestController public class MessageController { PostMapping(/sendPrivate) public String sendPrivate(RequestBody ChatMessage message) { boolean success ChatService.sendToUser(message.getToUserId(), message.getContent()); return success ? 已发送 : 目标用户不在线; } PostMapping(/broadcast) public String broadcast(RequestBody ChatMessage message) { ChatService.broadcast(message.getContent()); return 广播已触发; } }这样一个完整闭环就出来了浏览器A连上WebSocket浏览器B通过HTTP接口广播一条消息A立刻收到。私发同理。实际验证时我建议用两个浏览器窗口一个正常连ws另一个通过Postman或curl触发发送能直观看到消息的收发放结果。再提供一个你能直接在前端控制台测试的连法let ws new WebSocket(ws://localhost:8080/ws?userId1001); ws.onmessage function(event) { console.log(收到消息, event.data); }; ws.onopen function() { ws.send(JSON.stringify({type: broadcast, fromUserId: 1001, content: 大家好})); };前端这十几行代码够把整个链路从连接到广播验一遍。4. 常见问题与排查技巧实录4.1 私发消息总是发送失败先看目标用户在线表做私发遇到最多的问题就是消息发不出去但代码又不报错。第一反应不用去看网络先去看SessionManager里到底有没有目标用户的session。我自己的排查顺序是先打印在线用户数量和全部userId列表确认目标用户是否上线。如果没有就回到前端确认连接是否真的建立成功有时候是Denied权限安全限制导致WebSocket握手失败浏览器控制台不会报错但你实际没连上。再看连接参数。很多同学在代码里硬编码了fromUserId和toUserId测试时两个ID写反消息发给了自己或者发给了并不存在的用户排查了半天才发现是测试数据的问题。最后看生命周期。onOpen是否执行了onOpen里是否把session注册进SessionManager了。漏了注册这一个动作后面所有判断都是“目标不在线”。建议写代码的时候把SessionManager的操作日志打全上线、掉线、收到消息、发送成功分别打印线上排查的时候这些日志比什么都管用。4.2 广播卡死或延迟飙升多半是串行发送惹的祸广播高峰期消息发不出去问题大概率出在发送方式上。我早期写过一版广播用的是session.getBasicRemote().sendText循环发结果响应特别快但到了300个在线用户之后CPU就飙到80%。后来一分析问题就出在getBasicRemote是阻塞发送一个慢连接或者一个没收到TCP确认的客户端会卡住整个发送线程。解决方式有两个思路。一个是用session.getAsyncRemote().sendText异步发送调用后立即返回底层有线程池处理真正的写入吞吐量能提升不少。异步发送的问题是错误处理和顺序保证会复杂一些要在sendText的回调里记录失败日志。另一个思路是优化广播策略。比如合并消息短时间内的多条广播先合并成一条或者使用多线程分片每批处理500个session不要一个循环从1跑到10000。这都属于性能调优的经验但Demo阶段可以先了解不会踩太大坑。4.3 连接假死客户端显示在线实际消息发不过去WebSocket有一个经典的假死场景客户端断网比如电梯里手机信号断了TCP连接没有被正常关闭服务端完全不知道session还留在在线表里。这时候广播会往这个假死连接上写数据写了几次之后底层发现写不下去才开始抛异常但之前已经浪费了大量重传时间。处理假死只有一个标准方案心跳机制。客户端每隔30秒或60秒发送一个ping消息服务端收到后更新该用户的lastHeartbeat时间服务端每隔一段时间扫描一次找出超过一定时间没心跳的连接主动关闭并移出在线表。具体实现上可以用WebSocket自带的Ping/Pong控制帧也可以业务层自己约定一个ping消息格式。这个机制必须在设计阶段就考虑进去否则线上出一次假死问题排查起来非常痛苦。我在Demo里建议至少把lastHeartbeat字段加上后面接一个定时任务扫描不需要复杂设计就能杜绝大部分连接残留问题。4.4 部署在Nginx后面的连接被断掉很多同学本地测试一切正常一旦部署到服务器WebSocket连接稳定不了几秒就被断开。这类问题十有八九是反向代理没配置好。如果是Nginx必须设置Upgrade相关的Header把HTTP升级请求正确转发到后端。WebSocket是长连接代理默认的60秒超时不一定够用还需要设置较长的proxy_read_timeout。这些配置属于部署细节但真到生产环境绕不开。建议任何时候接入WebSocket之前先确认代理层已经支持协议升级。5. 从Demo走向生产还要补哪几节课5.1 消息可靠性从“尽力而为”到“至少一次”我在前面写的Demo本质是“消息发出去就算成功”这叫尽力而为的投递。但在很多真实业务里比如订单状态变更、支付结果通知消息绝对不能丢。要提升可靠性需要引入ACK确认机制客户端收到消息后回一个ack给服务端服务端发送消息时带一个唯一的消息ID如果一段时间内没收到ack就重发。重发可能导致客户端收到重复消息所以客户端要按消息ID去重。这套机制听起来复杂但它是一个可靠消息系统的地基。5.2 集群部署下私发先定位“用户在哪一台机器”单机部署是极其罕见的真实环境基本都是多台服务器负载均衡。如果只在一台机器的SessionManager里存在线连接A连接在节点1广播的消息发到了节点2节点2根本没有A的session消息自然丢失。集群的解法主要有三种。一种是粘性会话让同一个用户的WebSocket固定连到同一台节点配置简单但节点重启会掉线。第二种是用中间件做会话广播当节点1广播时通过发布订阅通道通知节点2、节点3让所有节点都参与发送每个节点只发自己本地的session。这个方案需要引入外部依赖。第三种是集中式会话存储比如把session句柄放到一个用户在线状态服务里但session对象本身不好跨进程传送所以实践中更多用的是第二种。这块属于架构层面的扩展但你应该知道私发和广播的难点从来都不是那几十行发送代码而是站在集群视角如何解决“用户连接分散”的问题。5.3 离线消息和消息补偿别把实时系统做成消息黑洞还有一类需求在聊天系统里特别常见用户发了一条私发对方下线了消息不能直接从服务器消失应该等对方上线后再补推。最简单的方式是发送失败时落库接收方建立WebSocket连接后服务端主动把离线期间未读的消息捞出来推过去。再讲究一点的可以引入消息队列来做削峰先推到队列再异步推送避免大量推送任务同时抢占线程资源。离线消息其实解决了两个问题一是消息不能丢二是接收方能知道“我不在的时间段里发生了什么”。加上这一步才算是一个能闭环的消息系统。最后一个实战体会如果你只是为了完成课程作业或者快速验证功能上面第3章的Demo已经完全够用。如果你正在做的是生产级项目我建议把第2章的会话设计反复想透再补上第4章和第5章里的心跳、确认、离线消息。从我的经验看实现私发和广播本身不难难的是把边界情况都处理掉比如用户掉线了、重复连接了、消息发送一半连接断开、广播时一条坏连接拖垮整个批次。这些才是真正拉开普通代码和稳定系统差距的地方。按我现在的习惯私发和广播从来不会在同一个方法里裸奔发送而是统一封装一套发送服务入口负责解析消息出口负责投递策略。先把这条路走顺了后续加任何新消息类型都会轻松很多。 
长河君 · 2026/10/13 12:32:54 · 阅读 1.2k Java WebSocket消息推送：私发与广播的实现与踩坑实践
做后端开发的人基本都会碰到一类需求给某个指定用户推送一条消息或者朝全平台在线用户喊一嗓子。前者是私发后者是广播在Java生态里这俩经常被放到一起做组合成一个消息推送模块。私发类比微信单聊广播类比直播间的全量公告看起来都是“把一段文本从服务器发到浏览器/客户端”但真要落地成代码连接管理、消息协议、线程安全、断线重连任何一环偷懒都会在线上给你颜色看。这篇文章不绕弯子直接梳理“Java消息私发和广播”从设计到实现的完整链路。我会把核心会话管理、点对点消息、全员广播的实现思路掰开揉碎附带可以直接抄走的Spring Boot WebSocket Demo最后聊聊我实际踩过的坑。适合刚接触WebSocket的Java后端也适合正在把手写Socket升级成稳定模块的同学。1. 这个需求本质在解决什么问题1.1 私发和广播的场景拆解私发和广播虽然都叫“发消息”但背后是两套完全不同的消息分发模型。私发要求精确投递A用户发给B用户服务器必须先知道B在哪然后单独走一条通路把消息塞给B。这里的关键是“寻址”就像寄快递必须有收件人地址不能扔到小区门口就完事。广播要求的是扇出一条消息要复制N份推给所有当前在线的人。这里的关键是“遍历与容错”只要有一个连接出问题不能把整批发送都卡死。真实业务里这俩往往会混合出现。比如一个客服系统用户发给客服的消息是私发系统维护公告全员推送是广播某个工单状态变更既要通知提交人又要通知处理人这就是一组私发。想清楚使用场景再去设计代码才不会把接口写拧巴。1.2 技术选型为什么我选了WebSocket而不是轮询聊Java消息推送不止一种路子。最简单的就是HTTP轮询前端每隔几秒发一次请求问“有没有新消息”。这种方式优点是实现门槛极低缺点是服务器压力大、消息实时性差高峰期一群人狂点会直接把接口打崩体验也很一般。还有一个方案是SSEServer-Sent Events服务器单向往浏览器推适合公告流、通知流但浏览器端API和长连接管理做起来总觉得不太顺手。我最终选择WebSocket核心就三点全双工通信、一次握手长期复用、消息实时性拉满。WebSocket的优势不仅体现在“能发消息”更关键的是它能同时承载上行和下行。用户在线状态、心跳保活、客户端主动上报这些后续要加的功能都能在同一条连接上扩展。你当然可以用纯Java写一套原生的Socket 多线程来管理连接但Spring Boot自带的WebSocket Starter已经把端点注册、握手握手、连接生命周期这些底层逻辑封装好了我们只需要关注业务本身。1.3 一开始就定好消息模型后面能少改很多代码新手写消息推送最容易犯的错是拿到需求直接写一个sendText(String message)把消息原样丢出去结果没过几天就发现不对劲没人知道这条消息到底是要私发还是广播也没人知道消息体里该放什么字段。我在动手写代码前一定先把消息协议定好。哪怕只是Demo也要考虑消息类型、目标用户、消息内容、发送时间这几个基础字段。前端收到消息后先看消息类型再做对应处理后端解析时也能快速判断该走私发逻辑还是广播逻辑。有了这层约定后面接入前端、测试、线上排错都会省力得多。2. 核心会话管理设计与原理解析2.1 用一张线程安全的在线表管理所有连接消息推送要解决的头号问题是“服务器怎么知道用户当前连在哪”。HTTP是无状态的一个请求处理完连接就断WebSocket不一样连接会一直留着。于是我们必须自己维护一张“在线连接表”把用户标识和对应的WebSocket会话对应起来。这张表在设计上有几个硬性要求第一必须线程安全。WebSocket服务端是并发模型每来一条连接就会有一个线程或异步任务在处理多个用户同时上线、下线、收发消息肯定会有并发读写这张表。用普通的HashMap扩容和数据覆盖都是雷我习惯用ConcurrentHashMap读多写少的场景下它表现很稳。第二key要稳定唯一。选择用户ID作为key最直接但要注意区分业务场景。比如同一用户开两个浏览器标签页就有两个WebSocket连接到底以哪个为准这要在业务层面定策略要么同一个用户只保留最新连接旧连接主动关闭避免一个用户收到重复消息要么支持一个key对应多个session发送时全session都发。大多数后台管理系统选前者简单不易乱。第三value不能只存session对象。session里带了很多底层信息但光有个session还不够我们通常还会维护一个lastHeartbeat字段用来做心跳检测和踢掉死连接。所以我更倾向于用一个简单封装对象作为value而不是直接存裸session。伪代码大概是这样的Component public class SessionManager { private final ConcurrentHashMapString, ClientSession onlineUsers new ConcurrentHashMap(); public void add(String userId, Session session) { ClientSession clientSession new ClientSession(userId, session, System.currentTimeMillis()); onlineUsers.put(userId, clientSession); } public ClientSession remove(String userId) { return onlineUsers.remove(userId); } public Session getSession(String userId) { ClientSession clientSession onlineUsers.get(userId); return clientSession null ? null : clientSession.getSession(); } public CollectionClientSession getAllSessions() { return onlineUsers.values(); } public boolean isOnline(String userId) { return onlineUsers.containsKey(userId); } }这段代码看着简单但它是整个私发和广播模块的地基。后面所有发送逻辑都要围绕这张表展开。2.2 私发为什么必须查表定位私发的本质就是根据接收方ID到在线连接表里找对应的WebSocket会话然后推送消息。public boolean sendToUser(String userId, String message) { Session session sessionManager.getSession(userId); if (session null || !session.isOpen()) { return false; } session.getBasicRemote().sendText(message); return true; }这段代码里有两个容易被忽略的细节。第一个是session.isOpen()的判断。客户端可能已经断网断电服务端还没感知到如果把一条消息写到一条已经关闭的连接上会抛异常。所以在发送前必须先确认会话仍然有效。第二个是发送方法的返回。实时消息系统里“消息没发出去”是常态接收方离线、连接抖动、服务端重启都会导致投递失败。私发必须给出成功或失败的结果方便上层决定是落离线消息还是给发送方一个“对方不在线”的提示。如果直接把sendText包在try-catch里然后忽略结果短期看不出问题长期一定会积累一堆“消息发了但用户没收到”的线上反馈。2.3 广播为什么不能直接遍历所有连接广播看起来更简单把在线连接表遍历一遍挨个调用sendText就完事。但真正的广播需要考虑三个问题。问题一连接断开的隔离。遍历过程中如果某个sessionsendText抛出异常我见过不少人直接catch住就继续这是可以的但有个隐藏前提——你必须确认这个异常的session只是被跳过而不会影响后续其他session的发送。如果错误处理不好一个脏连接就把整个广播批次拖垮。所以我建议每个session独立try-catch这个原则和批量任务处理里的隔离思路是一样的。问题二消息风暴。如果在线用户有一万广播一次就要调用一万次sendText每次都是一次网络IO和底层协议封包。业务上线之后如果广播频率高比如每分钟广播一次服务器CPU和带宽都会面临压力。这种场景下广播不只是“发出去就完事”还需要考虑是否要限制频率、是否要对大文本做压缩这都属于生产环境必须补的课。问题三并发遍历与实时上下线冲突。遍历getAllSessions()时正好有人掉线可能迭代到已关闭的session并触发异常有人上线也可能出现拷贝迭代时拿到的是旧快照。ConcurrentHashMap在设计上弱一致迭代时不会抛并发修改异常但你要明白这并不代表迭代结果绝对实时。一个稳妥的广播方法长这样public void broadcast(String message) { for (ClientSession clientSession : sessionManager.getAllSessions()) { Session session clientSession.getSession(); if (session null || !session.isOpen()) { continue; } try { session.getBasicRemote().sendText(message); } catch (Exception e) { // 记录日志但不要中断整个广播 } } }这里的关键是先检查连接状态再单独发发错了只记录日志不抛出这样一条废连接不会拖累其他人。2.4 并发染上线程安全踩的坑HashMap转的教训这个坑我是真实踩过的。早期写消息推送的时候图省事直接用了HashMapString, Session上线那天几十个人同时连上来服务器console刷了一屏ConcurrentModificationException。问题就出在多个WebSocket线程同时put、同时遍历、同时removeHashMap在扩容时遇到并发写内部链表直接乱掉。解决方式很简单要么用ConcurrentHashMap要么用Collections.synchronizedMap包裹再加手动同步。前者代码最干净也是我最常用的一种。不过ConcurrentHashMap也不是万能比如“先判断再放入”这种复合操作仍然存在竞态条件// 有并发问题因为containsKey和put不是原子的 if (!onlineUsers.containsKey(userId)) { onlineUsers.put(userId, session); }真正需要原子注册的时候可以用putIfAbsent方法ClientSession old onlineUsers.putIfAbsent(userId, clientSession); if (old ! null) { // 说明用户已存在做去重或顶号处理 }同时连接、同时心跳、同时掉线这种多线程场景在消息系统里是常态所以在线连接表里每一步操作都需要考虑并发语义。3. 实操过程用Java WebSocket跑通一个私发和广播的Demo3.1 工程结构与依赖准备我直接用Spring Boot来搭这个Demo版本用Spring Boot 2.7.x或者3.x都可以核心依赖只需要两个spring-boot-starter-web和spring-boot-starter-websocket。前者提供HTTP接口方便我们做一个触发消息的入口后者提供WebSocket端点支持。这一步有一个非常重要的配置如果你用的是Spring Boot内置容器需要注册ServerEndpointExporter这个bean它负责扫描所有标注了ServerEndpoint注解的类。有些同学照着教程写了端点却始终访问不到八成就是漏了这个配置。pom里加上依赖后最重要的配置类如下Configuration public class WebSocketConfig { Bean public ServerEndpointExporter serverEndpointExporter() { return new ServerEndpointExporter(); } }这个配置就一行但是少写一行整个WebSocket都起不来排查的时候很容易被忽略。3.2 定义消息结构私发和广播先做区分消息结构我在前面强调过这一步真的不能省。这里我定义一个简单的消息封装类用Jackson序列化和反序列化public class ChatMessage { private String type; // private: 私发, broadcast: 广播 private String fromUserId; private String toUserId; private String content; private long timestamp; // 省略getter/setter }type字段是整个协议的开关。收到一条消息后先解析JSON拿到type再判断是走私发逻辑还是广播逻辑。这个设计的可扩展性也很好后面想加“群组消息”可以在type里扩展比如group再补充一个groupId字段代码结构不用推倒重来。另外我会习惯给消息加一个timestamp虽然Demo阶段看起来没用但真正出问题的时候对日志、排查延迟、确认顺序都靠这个字段。3.3 WebSocket处理器核心的消息入口端点类是整个模块的心脏所有连接建立、消息到达、连接关闭都会触发这里的方法。Component ServerEndpoint(/ws) public class MessageEndpoint { private Session session; private String userId; OnOpen public void onOpen(Session session) { this.session session; this.userId session.getRequestParameterMap().getOrDefault(userId, unknown).get(0); SessionManager.getInstance().add(userId, session); System.out.println(用户 userId 上线当前在线 SessionManager.getInstance().getOnlineCount()); } OnMessage public void onMessage(String message, Session session) { ChatMessage msg JSON.parseObject(message, ChatMessage.class); if (private.equals(msg.getType())) { // 私发 sendPrivate(msg); } else if (broadcast.equals(msg.getType())) { // 广播 broadcastMsg(msg); } } OnClose public void onClose(Session session) { SessionManager.getInstance().remove(userId); } OnError public void onError(Session session, Throwable error) { SessionManager.getInstance().remove(userId); } }这里有几个设计点需要重点解释。第一类是userId从哪来。Demo里我直接放在连接URL上客户端连接时带上?userId1001这种参数。实际项目中一定要换成token鉴权让后端从令牌中解析出用户身份而不是让客户端自己上报一个ID否则任何一个人都可以伪造别人ID来收消息这是Chat应用的重大安全漏洞。第二是onClose和onError里都要清理会话。很多同学只写了OnClose结果网络异常断开时连接表里残留了一堆僵尸session发消息的时候对着一个已经死掉的session一直报错。记住onError的触发频率远比你想象的高。第三是消息入口要try-catch。虽然框架层面不允许异常外抛但如果你在onMessage期间抛出了未捕获异常WebSocket连接会被直接关闭。所以解析消息、发送消息都建议包一层异常处理只记录日志不要让底层的IO异常把整个端点拖挂。3.4 私发与广播的发送实现接入层收到消息后真正的动作落在发送逻辑上。我把这俩方法单独抽出来方便复用private void sendPrivate(ChatMessage msg) { Session targetSession SessionManager.getInstance().getSession(msg.getToUserId()); if (targetSession null || !targetSession.isOpen()) { System.out.println(用户 msg.getToUserId() 不在线); return; } try { targetSession.getBasicRemote().sendText(JSON.toJSONString(msg)); } catch (IOException e) { System.err.println(私发消息失败 e.getMessage()); } } private void broadcastMsg(ChatMessage msg) { String messageText JSON.toJSONString(msg); for (ClientSession clientSession : SessionManager.getInstance().getAllSessions()) { if (clientSession.getSession().isOpen()) { try { clientSession.getSession().getBasicRemote().sendText(messageText); } catch (IOException e) { System.err.println(广播发送失败用户 clientSession.getUserId()); } } } }如果你把代码跑起来会发现一个有意思的问题私发的时候如果接收方不在线这段代码只是打一行日志就什么都不做了。真实业务里这显然不够所以生产环境中这里会触发离线消息存储把消息存进数据库或者消息队列等接收方上线后再拉取。私发和离线消息是一对孪生兄弟别只看在线发送这一层。3.5 加一个HTTP触发入口验证起来就方便了WebSocket是客户端主动连接后收消息但开发联调时总不可能每次都在另一个客户端里输入JSON去触发广播。我习惯在Demo里顺带提供两个HTTP接口用来模拟“管理端发私发”和“服务端发广播”。RestController public class MessageController { PostMapping(/sendPrivate) public String sendPrivate(RequestBody ChatMessage message) { boolean success ChatService.sendToUser(message.getToUserId(), message.getContent()); return success ? 已发送 : 目标用户不在线; } PostMapping(/broadcast) public String broadcast(RequestBody ChatMessage message) { ChatService.broadcast(message.getContent()); return 广播已触发; } }这样一个完整闭环就出来了浏览器A连上WebSocket浏览器B通过HTTP接口广播一条消息A立刻收到。私发同理。实际验证时我建议用两个浏览器窗口一个正常连ws另一个通过Postman或curl触发发送能直观看到消息的收发放结果。再提供一个你能直接在前端控制台测试的连法let ws new WebSocket(ws://localhost:8080/ws?userId1001); ws.onmessage function(event) { console.log(收到消息, event.data); }; ws.onopen function() { ws.send(JSON.stringify({type: broadcast, fromUserId: 1001, content: 大家好})); };前端这十几行代码够把整个链路从连接到广播验一遍。4. 常见问题与排查技巧实录4.1 私发消息总是发送失败先看目标用户在线表做私发遇到最多的问题就是消息发不出去但代码又不报错。第一反应不用去看网络先去看SessionManager里到底有没有目标用户的session。我自己的排查顺序是先打印在线用户数量和全部userId列表确认目标用户是否上线。如果没有就回到前端确认连接是否真的建立成功有时候是Denied权限安全限制导致WebSocket握手失败浏览器控制台不会报错但你实际没连上。再看连接参数。很多同学在代码里硬编码了fromUserId和toUserId测试时两个ID写反消息发给了自己或者发给了并不存在的用户排查了半天才发现是测试数据的问题。最后看生命周期。onOpen是否执行了onOpen里是否把session注册进SessionManager了。漏了注册这一个动作后面所有判断都是“目标不在线”。建议写代码的时候把SessionManager的操作日志打全上线、掉线、收到消息、发送成功分别打印线上排查的时候这些日志比什么都管用。4.2 广播卡死或延迟飙升多半是串行发送惹的祸广播高峰期消息发不出去问题大概率出在发送方式上。我早期写过一版广播用的是session.getBasicRemote().sendText循环发结果响应特别快但到了300个在线用户之后CPU就飙到80%。后来一分析问题就出在getBasicRemote是阻塞发送一个慢连接或者一个没收到TCP确认的客户端会卡住整个发送线程。解决方式有两个思路。一个是用session.getAsyncRemote().sendText异步发送调用后立即返回底层有线程池处理真正的写入吞吐量能提升不少。异步发送的问题是错误处理和顺序保证会复杂一些要在sendText的回调里记录失败日志。另一个思路是优化广播策略。比如合并消息短时间内的多条广播先合并成一条或者使用多线程分片每批处理500个session不要一个循环从1跑到10000。这都属于性能调优的经验但Demo阶段可以先了解不会踩太大坑。4.3 连接假死客户端显示在线实际消息发不过去WebSocket有一个经典的假死场景客户端断网比如电梯里手机信号断了TCP连接没有被正常关闭服务端完全不知道session还留在在线表里。这时候广播会往这个假死连接上写数据写了几次之后底层发现写不下去才开始抛异常但之前已经浪费了大量重传时间。处理假死只有一个标准方案心跳机制。客户端每隔30秒或60秒发送一个ping消息服务端收到后更新该用户的lastHeartbeat时间服务端每隔一段时间扫描一次找出超过一定时间没心跳的连接主动关闭并移出在线表。具体实现上可以用WebSocket自带的Ping/Pong控制帧也可以业务层自己约定一个ping消息格式。这个机制必须在设计阶段就考虑进去否则线上出一次假死问题排查起来非常痛苦。我在Demo里建议至少把lastHeartbeat字段加上后面接一个定时任务扫描不需要复杂设计就能杜绝大部分连接残留问题。4.4 部署在Nginx后面的连接被断掉很多同学本地测试一切正常一旦部署到服务器WebSocket连接稳定不了几秒就被断开。这类问题十有八九是反向代理没配置好。如果是Nginx必须设置Upgrade相关的Header把HTTP升级请求正确转发到后端。WebSocket是长连接代理默认的60秒超时不一定够用还需要设置较长的proxy_read_timeout。这些配置属于部署细节但真到生产环境绕不开。建议任何时候接入WebSocket之前先确认代理层已经支持协议升级。5. 从Demo走向生产还要补哪几节课5.1 消息可靠性从“尽力而为”到“至少一次”我在前面写的Demo本质是“消息发出去就算成功”这叫尽力而为的投递。但在很多真实业务里比如订单状态变更、支付结果通知消息绝对不能丢。要提升可靠性需要引入ACK确认机制客户端收到消息后回一个ack给服务端服务端发送消息时带一个唯一的消息ID如果一段时间内没收到ack就重发。重发可能导致客户端收到重复消息所以客户端要按消息ID去重。这套机制听起来复杂但它是一个可靠消息系统的地基。5.2 集群部署下私发先定位“用户在哪一台机器”单机部署是极其罕见的真实环境基本都是多台服务器负载均衡。如果只在一台机器的SessionManager里存在线连接A连接在节点1广播的消息发到了节点2节点2根本没有A的session消息自然丢失。集群的解法主要有三种。一种是粘性会话让同一个用户的WebSocket固定连到同一台节点配置简单但节点重启会掉线。第二种是用中间件做会话广播当节点1广播时通过发布订阅通道通知节点2、节点3让所有节点都参与发送每个节点只发自己本地的session。这个方案需要引入外部依赖。第三种是集中式会话存储比如把session句柄放到一个用户在线状态服务里但session对象本身不好跨进程传送所以实践中更多用的是第二种。这块属于架构层面的扩展但你应该知道私发和广播的难点从来都不是那几十行发送代码而是站在集群视角如何解决“用户连接分散”的问题。5.3 离线消息和消息补偿别把实时系统做成消息黑洞还有一类需求在聊天系统里特别常见用户发了一条私发对方下线了消息不能直接从服务器消失应该等对方上线后再补推。最简单的方式是发送失败时落库接收方建立WebSocket连接后服务端主动把离线期间未读的消息捞出来推过去。再讲究一点的可以引入消息队列来做削峰先推到队列再异步推送避免大量推送任务同时抢占线程资源。离线消息其实解决了两个问题一是消息不能丢二是接收方能知道“我不在的时间段里发生了什么”。加上这一步才算是一个能闭环的消息系统。最后一个实战体会如果你只是为了完成课程作业或者快速验证功能上面第3章的Demo已经完全够用。如果你正在做的是生产级项目我建议把第2章的会话设计反复想透再补上第4章和第5章里的心跳、确认、离线消息。从我的经验看实现私发和广播本身不难难的是把边界情况都处理掉比如用户掉线了、重复连接了、消息发送一半连接断开、广播时一条坏连接拖垮整个批次。这些才是真正拉开普通代码和稳定系统差距的地方。按我现在的习惯私发和广播从来不会在同一个方法里裸奔发送而是统一封装一套发送服务入口负责解析消息出口负责投递策略。先把这条路走顺了后续加任何新消息类型都会轻松很多。