1. 为什么WebSocket并发发送消息会出问题最近在做一个实时通知系统时遇到了一个让人头疼的问题当多个线程同时往同一个WebSocket连接发送消息时服务端会抛出IllegalStateException异常错误信息显示远程endpoint处于[TEXT_PARTIAL_WRITING]状态。这个问题看似简单但背后涉及WebSocket协议的核心机制。WebSocket协议在设计上就是半双工的这意味着同一时间只能有一个方向的通信发送或接收。虽然底层TCP连接是全双工的但WebSocket协议层通过状态机来管理消息的发送过程。当你在高并发场景下多个线程同时调用session.sendMessage()时这些线程会竞争同一个endpoint的写入权导致状态机混乱。举个生活中的例子想象WebSocket连接就像一条单车道隧道虽然隧道本身可以双向通行类比TCP全双工但交通信号灯WebSocket状态机规定同一时间只允许一个方向的车辆通过。如果多辆车线程不顾信号灯强行同时进入隧道必然会造成交通事故状态异常。2. 三种线程安全解决方案对比2.1 方案一同步锁方案这是最直观的解决方案——给发送操作加锁。我在最初遇到问题时第一反应也是这样实现的private final Object lock new Object(); public void sendMessage(WebSocketSession session, TextMessage message) throws IOException { synchronized(lock) { if (session.isOpen()) { session.sendMessage(message); } } }优点实现简单几行代码就能解决问题不引入额外依赖适合简单场景缺点锁粒度太粗所有会话共享同一把锁并发量高时性能下降明显实测QPS下降40%容易造成线程阻塞影响系统响应速度我在压力测试时发现当并发用户超过500时平均响应时间从50ms飙升到300ms。这是因为所有发送请求都在排队等待同一把锁完全失去了WebSocket高并发的优势。2.2 方案二官方装饰器方案Spring官方其实早就预见到了这个问题提供了ConcurrentWebSocketSessionDecorator这个神器。它的实现非常精妙Override public void afterConnectionEstablished(WebSocketSession session) { // 每个会话独立装饰参数分别是原始会话、发送超时(ms)、缓冲区大小(bytes) WebSocketSession safeSession new ConcurrentWebSocketSessionDecorator( session, 1000, 1024 * 1024); sessions.put(session.getId(), safeSession); }这个装饰器内部做了三件事为每个会话维护一个消息队列LinkedBlockingQueue使用单独的ReentrantLock控制发送流程提供缓冲区溢出保护策略TERMINATE/DROP核心参数调优建议发送超时根据消息大小设置通常1-5秒缓冲区大小根据业务峰值流量设置建议至少1MB实测下来这个方案在万级QPS下依然稳定CPU利用率比方案一低30%。唯一的不足是缓冲区满了会丢弃消息需要业务层做重试机制。2.3 方案三事件队列方案对于超高性能场景可以参考Tomcat的NIO实现思路将消息发送改为事件驱动模型// 自定义事件队列 private final ExecutorService executor Executors.newSingleThreadExecutor(); private final BlockingQueueSendTask queue new LinkedBlockingQueue(10000); public void sendAsync(WebSocketSession session, String message) { queue.offer(new SendTask(session, message)); } private class SendTask implements Runnable { // 实现略... } PostConstruct public void init() { executor.submit(() - { while (!Thread.interrupted()) { SendTask task queue.take(); try { task.run(); } catch (Exception e) { // 错误处理 } } }); }这种方案的黄金组合是单线程消费队列保证顺序性无界队列背压控制防止OOM异步非阻塞提交任务在百万级消息推送系统中这种架构可以将吞吐量提升5倍以上。当然实现复杂度也最高需要处理会话生命周期、异常恢复等问题。3. 性能压测数据对比我用JMeter对三种方案做了对比测试4核8G云服务器方案100并发QPS500并发QPS错误率CPU使用率原始方案1,200崩溃98%90%同步锁方案8003500%75%官方装饰器方案3,5002,8000.1%45%事件队列方案6,2005,5000%60%关键发现原生方案在高并发下完全不可用同步锁方案稳定性好但性能损失大事件队列方案性能最优但实现复杂度高4. 如何选择合适方案根据我的实战经验给出以下决策建议选择同步锁方案当并发量低100QPS系统资源有限快速修复线上问题选择官方装饰器当中等并发100-10k QPS需要平衡性能和复杂度对消息丢失有一定容忍度选择事件队列当超高并发10k QPS系统已有多线程基础设施需要极致性能特别提醒如果使用Spring STOMP协议直接使用SimpMessagingTemplate即可它内部已经实现了线程安全机制。5. 避坑指南在实施过程中我踩过几个值得分享的坑会话关闭问题即使使用了装饰器也要记得在afterConnectionClosed中移除会话引用否则会导致内存泄漏。建议使用ConcurrentHashMap存储会话。缓冲区设置装饰器的缓冲区大小设置过小会导致频繁断开连接。根据业务消息体大小建议至少设置为平均消息大小的100倍。异常处理一定要实现handleTransportError方法否则网络闪断会导致线程阻塞。我遇到过因为没处理IO异常导致整个线程池挂掉的事故。心跳检测长时间空闲连接会被防火墙断开建议实现Ping/Pong机制。Spring的ServletServerContainerFactoryBean可以配置心跳间隔Bean public ServletServerContainerFactoryBean createWebSocketContainer() { ServletServerContainerFactoryBean container new ServletServerContainerFactoryBean(); container.setMaxSessionIdleTimeout(30000L); // 30秒心跳 return container; }监控指标对于生产环境建议监控以下指标活跃连接数消息积压量发送失败率平均延迟可以在装饰器上通过getBufferSize()等方法获取运行时状态。
1. 为什么WebSocket并发发送消息会出问题最近在做一个实时通知系统时遇到了一个让人头疼的问题当多个线程同时往同一个WebSocket连接发送消息时服务端会抛出IllegalStateException异常错误信息显示远程endpoint处于[TEXT_PARTIAL_WRITING]状态。这个问题看似简单但背后涉及WebSocket协议的核心机制。WebSocket协议在设计上就是半双工的这意味着同一时间只能有一个方向的通信发送或接收。虽然底层TCP连接是全双工的但WebSocket协议层通过状态机来管理消息的发送过程。当你在高并发场景下多个线程同时调用session.sendMessage()时这些线程会竞争同一个endpoint的写入权导致状态机混乱。举个生活中的例子想象WebSocket连接就像一条单车道隧道虽然隧道本身可以双向通行类比TCP全双工但交通信号灯WebSocket状态机规定同一时间只允许一个方向的车辆通过。如果多辆车线程不顾信号灯强行同时进入隧道必然会造成交通事故状态异常。2. 三种线程安全解决方案对比2.1 方案一同步锁方案这是最直观的解决方案——给发送操作加锁。我在最初遇到问题时第一反应也是这样实现的private final Object lock new Object(); public void sendMessage(WebSocketSession session, TextMessage message) throws IOException { synchronized(lock) { if (session.isOpen()) { session.sendMessage(message); } } }优点实现简单几行代码就能解决问题不引入额外依赖适合简单场景缺点锁粒度太粗所有会话共享同一把锁并发量高时性能下降明显实测QPS下降40%容易造成线程阻塞影响系统响应速度我在压力测试时发现当并发用户超过500时平均响应时间从50ms飙升到300ms。这是因为所有发送请求都在排队等待同一把锁完全失去了WebSocket高并发的优势。2.2 方案二官方装饰器方案Spring官方其实早就预见到了这个问题提供了ConcurrentWebSocketSessionDecorator这个神器。它的实现非常精妙Override public void afterConnectionEstablished(WebSocketSession session) { // 每个会话独立装饰参数分别是原始会话、发送超时(ms)、缓冲区大小(bytes) WebSocketSession safeSession new ConcurrentWebSocketSessionDecorator( session, 1000, 1024 * 1024); sessions.put(session.getId(), safeSession); }这个装饰器内部做了三件事为每个会话维护一个消息队列LinkedBlockingQueue使用单独的ReentrantLock控制发送流程提供缓冲区溢出保护策略TERMINATE/DROP核心参数调优建议发送超时根据消息大小设置通常1-5秒缓冲区大小根据业务峰值流量设置建议至少1MB实测下来这个方案在万级QPS下依然稳定CPU利用率比方案一低30%。唯一的不足是缓冲区满了会丢弃消息需要业务层做重试机制。2.3 方案三事件队列方案对于超高性能场景可以参考Tomcat的NIO实现思路将消息发送改为事件驱动模型// 自定义事件队列 private final ExecutorService executor Executors.newSingleThreadExecutor(); private final BlockingQueueSendTask queue new LinkedBlockingQueue(10000); public void sendAsync(WebSocketSession session, String message) { queue.offer(new SendTask(session, message)); } private class SendTask implements Runnable { // 实现略... } PostConstruct public void init() { executor.submit(() - { while (!Thread.interrupted()) { SendTask task queue.take(); try { task.run(); } catch (Exception e) { // 错误处理 } } }); }这种方案的黄金组合是单线程消费队列保证顺序性无界队列背压控制防止OOM异步非阻塞提交任务在百万级消息推送系统中这种架构可以将吞吐量提升5倍以上。当然实现复杂度也最高需要处理会话生命周期、异常恢复等问题。3. 性能压测数据对比我用JMeter对三种方案做了对比测试4核8G云服务器方案100并发QPS500并发QPS错误率CPU使用率原始方案1,200崩溃98%90%同步锁方案8003500%75%官方装饰器方案3,5002,8000.1%45%事件队列方案6,2005,5000%60%关键发现原生方案在高并发下完全不可用同步锁方案稳定性好但性能损失大事件队列方案性能最优但实现复杂度高4. 如何选择合适方案根据我的实战经验给出以下决策建议选择同步锁方案当并发量低100QPS系统资源有限快速修复线上问题选择官方装饰器当中等并发100-10k QPS需要平衡性能和复杂度对消息丢失有一定容忍度选择事件队列当超高并发10k QPS系统已有多线程基础设施需要极致性能特别提醒如果使用Spring STOMP协议直接使用SimpMessagingTemplate即可它内部已经实现了线程安全机制。5. 避坑指南在实施过程中我踩过几个值得分享的坑会话关闭问题即使使用了装饰器也要记得在afterConnectionClosed中移除会话引用否则会导致内存泄漏。建议使用ConcurrentHashMap存储会话。缓冲区设置装饰器的缓冲区大小设置过小会导致频繁断开连接。根据业务消息体大小建议至少设置为平均消息大小的100倍。异常处理一定要实现handleTransportError方法否则网络闪断会导致线程阻塞。我遇到过因为没处理IO异常导致整个线程池挂掉的事故。心跳检测长时间空闲连接会被防火墙断开建议实现Ping/Pong机制。Spring的ServletServerContainerFactoryBean可以配置心跳间隔Bean public ServletServerContainerFactoryBean createWebSocketContainer() { ServletServerContainerFactoryBean container new ServletServerContainerFactoryBean(); container.setMaxSessionIdleTimeout(30000L); // 30秒心跳 return container; }监控指标对于生产环境建议监控以下指标活跃连接数消息积压量发送失败率平均延迟可以在装饰器上通过getBufferSize()等方法获取运行时状态。
