最近在做一个智能客服项目客户对实时性和并发量要求都很高。传统的基于HTTP轮询的方案延迟高、服务器压力大显然无法满足需求。经过一番技术选型和实践我们最终采用Java WebSocket作为通信核心结合AI模型搭建了一套高并发、低延迟的智能客服系统。今天就把整个实战过程和一些优化心得记录下来希望能给有类似需求的同学一些参考。1. 为什么选择 WebSocket聊聊传统客服的痛点在项目初期我们复盘了传统客服系统常见的几个问题响应延迟高这是HTTP轮询的“原罪”。客户端需要不断向服务器发送“你那边有我的新消息吗”的请求大部分请求都是无效的白白浪费了网络带宽和服务器资源消息从发出到被客户端感知延迟通常在秒级。服务器压力大想象一下成千上万的用户每隔几秒就发起一次HTTP请求即使没有新消息服务器也要处理这些连接、解析请求、返回空响应。这对服务器资源是极大的消耗。扩展性差基于HTTP短连接的会话状态维护比较麻烦通常依赖于Session或Token在分布式环境下需要额外的方案如Redis来共享状态增加了系统复杂度。双向通信不便HTTP是“请求-响应”模型服务器无法主动向客户端推送消息。虽然可以用长轮询或Server-Sent Events (SSE) 模拟但都不如WebSocket原生支持双向通信来得优雅和高效。相比之下WebSocket协议在建立连接后就提供了一个全双工的通信通道。客户端和服务器可以随时互发消息连接是持久的避免了频繁的握手和头部开销。这对于需要实时交互的客服场景简直是量身定做。2. 核心架构设计与技术栈我们的系统架构可以概括为以下几个核心部分WebSocket 网关层使用Java实现WebSocket服务端负责维护与所有客户端的持久连接处理连接建立、消息接收和推送。业务处理层接收来自WebSocket层的用户消息进行基本的校验和预处理。AI 引擎服务这是系统的“大脑”。我们通过RPC或HTTP调用一个独立的AI服务。这个服务内部会进行自然语言处理NLP例如意图识别、实体抽取然后调用预训练的语言模型如企业内部微调过的模型或大模型API生成回复。消息队列 (MQ)这是解耦和削峰填谷的关键。当大量用户同时提问时直接将请求丢给AI服务可能会导致服务雪崩。我们将用户请求包装成消息发送到消息队列如RabbitMQ, Kafka由后端的AI Worker异步消费处理处理完毕后再将结果通过WebSocket连接推回给对应用户。会话与状态管理使用Redis来存储用户会话上下文。因为AI模型通常需要历史对话记录来理解当前问题所以每次对话都需要带上最近几轮的上下文。Redis的高性能非常适合这种场景。3. WebSocket 服务端核心实现我们使用javax.websocketAPIJSR 356来实现服务端它非常简洁。下面是一个最核心的端点类示例import javax.websocket.*; import javax.websocket.server.ServerEndpoint; import java.io.IOException; import java.util.concurrent.ConcurrentHashMap; ServerEndpoint(/chat) // 指定WebSocket访问路径 public class CustomerServiceEndpoint { // 使用线程安全的Map来管理在线会话Key通常为用户ID private static ConcurrentHashMapString, Session onlineSessions new ConcurrentHashMap(); /** * 连接建立成功时触发 */ OnOpen public void onOpen(Session session, PathParam(userId) String userId) { onlineSessions.put(userId, session); System.out.println(用户[ userId ] 连接成功当前在线人数: onlineSessions.size()); // 可以在这里发送欢迎消息或初始化会话 } /** * 收到客户端消息时触发 */ OnMessage public void onMessage(String message, Session session, PathParam(userId) String userId) { System.out.println(收到来自用户[ userId ] 的消息: message); // 1. 基础校验如敏感词过滤 if (!SecurityFilter.check(message)) { sendMessageToUser(userId, 您的问题包含敏感信息请重新输入。); return; } // 2. 将用户消息、用户ID、当前会话ID等封装成任务 ChatTask task new ChatTask(userId, session.getId(), message); // 3. 将任务投递到消息队列进行异步处理避免阻塞WebSocket线程 MessageQueueProducer.sendTask(task); // 4. 可以立即返回一个“正在思考”的提示 sendMessageToUser(userId, AI客服正在思考中请稍候...); } /** * 连接关闭时触发 */ OnClose public void onClose(Session session, PathParam(userId) String userId) { onlineSessions.remove(userId); System.out.println(用户[ userId ] 断开连接当前在线人数: onlineSessions.size()); // 清理该用户的Redis会话上下文等资源 } /** * 发生错误时触发 */ OnError public void onError(Session session, Throwable error, PathParam(userId) String userId) { System.err.println(用户[ userId ] 的连接发生错误: error.getMessage()); error.printStackTrace(); // 通常在这里记录日志并可能关闭有问题的会话 } /** * 向指定用户发送消息工具方法 */ public static void sendMessageToUser(String userId, String message) { Session session onlineSessions.get(userId); if (session ! null session.isOpen()) { try { // getBasicRemote() 是同步发送getAsyncRemote() 是异步发送高并发下推荐异步 session.getAsyncRemote().sendText(message); } catch (Exception e) { System.err.println(向用户[ userId ] 发送消息失败: e.getMessage()); } } } }关键点说明ServerEndpoint声明这是一个WebSocket端点。OnOpen,OnMessage,OnClose,OnError对应生命周期的注解。异步发送session.getAsyncRemote().sendText()是非阻塞的在高并发下比同步发送性能好得多。会话管理我们用一个静态Map来维护在生产环境中单机Map无法满足分布式部署需要替换为Redis等集中式存储来跟踪用户与WebSocket服务器实例的映射关系。4. 集成AI模型与异步处理AI服务通常比较耗时几百毫秒到几秒绝不能阻塞WebSocket的IO线程。我们的流程如下消息队列解耦MessageQueueProducer.sendTask(task)将任务发送到如RabbitMQ的队列中。AI Worker消费后台有多个AI Worker进程监听队列。Worker收到任务后从Redis中取出该用户的对话历史。将历史记录和当前问题组合调用AI模型API可能是内部的NLP服务也可能是OpenAI等大模型的接口。获得AI生成的回复。将本次问答存入Redis更新对话历史。调用CustomerServiceEndpoint.sendMessageToUser将回复推送给用户。上下文管理Redis中为每个用户存储一个列表保存最近N轮对话这是保证AI回复连贯性的关键。// AI Worker 伪代码示例 public class AIWorker { public void handleTask(ChatTask task) { String userId task.getUserId(); String question task.getMessage(); // 1. 从Redis获取历史对话 ListString history redisClient.lrange(chat:history: userId, 0, -1); // 2. 构建Prompt调用AI服务 String prompt buildPrompt(history, question); String aiResponse callAIService(prompt); // 可能是HTTP请求 // 3. 保存新的对话到历史记录控制长度比如只保留最近10轮 redisClient.lpush(chat:history: userId, 用户: question, AI: aiResponse); redisClient.ltrim(chat:history: userId, 0, 19); // 保留最多10轮20条消息 // 4. 通过WebSocket推送回复 CustomerServiceEndpoint.sendMessageToUser(userId, aiResponse); } }5. 性能优化与安全考量性能优化连接池与线程池WebSocket服务器本身如Tomcat、Undertow需要配置合适的线程池如Executor来处理IO事件。避免使用默认配置导致并发上不去。消息压缩对于较长的AI回复文本可以在WebSocket层面开启permessage-deflate扩展进行压缩传输节省带宽。序列化优化如果传输的不是纯文本而是复杂的JSON对象可以考虑使用Protobuf或MsgPack等二进制序列化方案比JSON更省空间、解析更快。心跳保活WebSocket连接可能因为防火墙、代理等原因意外断开。需要实现心跳机制Ping/Pong定期检查连接健康度及时清理死连接。安全性考量使用 WSS生产环境必须使用wss://WebSocket Secure即基于TLS/SSL加密的WebSocket防止中间人攻击和消息窃听。身份认证不要在URL参数中明文传递用户ID。应该在连接建立时通过标准的认证流程如校验Token来绑定用户身份。可以在OnOpen方法中实现。输入校验与过滤防注入虽然AI模型处理自然语言但传给AI服务前仍需对用户输入进行基本的SQL注入、脚本注入XSS检查。敏感词过滤建立敏感词库对用户输入和AI输出进行双重过滤确保合规性。这可以在WebSocket层和AI服务返回层都做一遍。6. 生产环境避坑指南在实际部署和运维中我们踩过一些坑这里分享给大家连接泄漏这是最常见的问题。务必在OnClose和OnError方法中做好资源清理工作从在线会话Map中移除并关闭可能未正确关闭的Session。同时要配置服务器如Netty的连接超时和空闲超时时间。消息丢失与重复消费使用消息队列时要确保消息的可靠性。我们为每个ChatTask设置了唯一IDAI Worker处理完成后在推送消息前在Redis中做一个简单的“已处理”标记设置一个短时间的key防止网络重试等原因导致的消息重复推送。对于消息丢失要确保MQ的持久化配置正确。内存溢出ConcurrentHashMap如果存储了大量Session对象可能引发OOM。除了前面提到的分布式会话管理还要注意Session对象本身可能关联的缓冲区。定期监控堆内存使用情况。集群部署难题单机WebSocket服务器有容量上限。要支持水平扩展就需要解决“会话路由”问题。即用户A连接到服务器1但AI处理完成后需要把消息推送给服务器1上的那个连接。我们引入了Redis Pub/Sub或专门的网关如Nginx的ip_hash或使用Spring Session等方案来维护用户与服务器实例的映射关系。AI服务超时与降级AI服务可能不稳定或响应慢。必须设置合理的超时时间如5秒并在超时或失败时返回一个友好的降级回复如“网络开小差了请稍后再试”或转向人工客服队列而不是让用户一直等待。7. 总结与展望通过这套基于Java WebSocket和AI模型的架构我们成功构建了一个响应迅速、能够处理高并发的智能客服系统。WebSocket解决了实时通信的瓶颈消息队列和异步处理保证了系统的稳定性和扩展性AI模型则提供了核心的智能交互能力。当然系统还有很大的优化空间AI模型升级可以尝试更先进的对话模型或者针对垂直领域进行精细微调以提供更准确、更专业的回答。架构演进当用户量进一步暴涨可以考虑将WebSocket网关层完全独立出来使用Netty等高性能框架自研并引入服务发现、负载均衡形成更彻底的分布式架构。功能丰富增加文件传输如图片、文档、富文本消息、对话评价、数据分析和可视化看板等功能让系统更加完善。构建这样一个系统技术选型只是第一步更重要的是在设计和开发过程中时刻考虑性能、可靠性和安全性。希望这篇笔记能为你提供一些可行的思路。如果你也在做类似的项目欢迎一起交流探讨。