最近在做一个电商小程序项目其中智能客服模块的实时性要求非常高。初期我们采用了传统的HTTP轮询方案结果在高并发活动期间服务器直接“躺平”了。经过一番重构我们最终基于WebSocket和消息队列搭建了一套新系统效果显著。今天就来分享一下从踩坑到填坑的全过程希望能给有类似需求的同学一些参考。1. 为什么轮询方案在高并发下会崩项目初期为了快速上线客服消息推送采用了最简单的HTTP短轮询。客户端每隔2-3秒就向服务器发起一次请求询问“有新消息吗”。这个方案在小流量时没问题但一到促销活动问题就全暴露出来了。连接数爆炸假设有1万用户同时在线每3秒轮询一次服务器每秒需要处理超过3000次HTTP请求。这还只是“询问”的请求不包括真正的业务请求。大量的连接建立和销毁消耗了服务器大量的CPU和内存资源。消息延迟高轮询间隔是固定的比如3秒。这意味着用户发送消息后最坏情况下对方需要等待一个完整的轮询周期3秒加上网络传输时间才能收到实时性很差用户体验大打折扣。无效请求多大部分轮询请求可能超过95%的响应都是“没有新消息”。这造成了巨大的网络带宽和服务器计算资源的浪费。服务器压力不均无论是否有消息请求都会均匀地打到服务器上无法根据实际业务负载进行动态调整。正是这些痛点迫使我们寻找更优的实时通信方案。2. 技术选型为什么是WebSocket MQ我们主要对比了三种主流方案长轮询、SSE和WebSocket。长轮询算是短轮询的改良版。客户端发起请求后服务器会“hold住”连接直到有数据或超时才返回。虽然减少了无效请求但每个请求仍然需要完整的HTTP生命周期连接管理复杂且服务器hold连接同样消耗资源。SSE服务器向浏览器单向推送基于HTTP协议实现简单。但它有两个硬伤一是单向通信服务器-客户端客服场景需要双向二是浏览器兼容性主要是IE系列不如WebSocket。WebSocket基于TCP的全双工通信协议。一次握手长久连接双向实时通信。这完美契合了客服场景“用户随时发客服随时回”的需求。它从根本上解决了轮询带来的连接开销和延迟问题。但是单靠WebSocket还不够。当数万甚至数十万连接同时在线时单机WebSocket服务肯定会成为瓶颈。因此我们引入了消息队列。它的作用是解耦和削峰填谷。将消息的“生产”用户发送和“消费”推送给客服/用户过程分离。在高并发写入时消息先进入队列后端服务可以按照自己的能力匀速消费避免了流量洪峰直接冲垮服务。所以WebSocket MQ的组合就成了我们的最终选择WebSocket解决实时双向通信MQ解决高并发下的可靠性与系统解耦。3. 核心实现分层架构与关键代码3.1 分层架构设计我们将系统划分为三层职责清晰便于扩展[ 客户端 ] --WebSocket-- [ 接入层 (WS-Gateway) ] | | (RPC / HTTP) v [ 逻辑层 (Message Service) ] | | (生产/消费消息) v [ 存储层 (Redis MySQL MQ) ]接入层由一组无状态的WebSocket网关服务器构成。它只负责维护海量的客户端长连接、解析基础协议、转发消息到逻辑层。它本身不处理业务逻辑因此可以轻松水平扩展。逻辑层处理核心业务逻辑的服务。包括消息的持久化、未读计数、用户与客服的匹配逻辑、敏感词过滤、推送逻辑等。它从MQ消费消息并决定要推送给哪个网关的哪个连接。存储层Redis缓存用户会话列表、未读消息数、在线状态、以及用作分布式锁。MySQL持久化存储所有消息记录、用户信息、客服信息等。MQ我们选用RabbitMQ它的队列模型和可靠性很适合。负责缓冲用户发送的消息以及逻辑层需要广播或推送的消息。3.2 WebSocket网关核心代码片段接入层网关使用Node.js的ws库实现因为它轻量且高效。// websocket-gateway.js const WebSocket require(ws); const { v4: uuidv4 } require(uuid); const redisClient require(./redis-client); const wss new WebSocket.Server({ port: 8080 }); const onlineClients new Map(); // 内存中维护连接映射userId - WebSocket wss.on(connection, async (ws, request) { // 1. 连接建立进行身份认证这里简化实际从token解析 const urlParams new URL(request.url, ws://${request.headers.host}); const token urlParams.searchParams.get(token); let userId; try { const userInfo await authToken(token); // 验证token获取用户ID userId userInfo.id; } catch (err) { ws.close(1008, Authentication failed); return; } // 2. 生成唯一连接ID并存储映射关系 const connectionId uuidv4(); ws.connectionId connectionId; onlineClients.set(userId, ws); // 3. 在Redis中记录用户在线状态和网关节点信息 await redisClient.hset(user:online:${userId}, gateway, gateway-node-1, connectionId, connectionId); await redisClient.expire(user:online:${userId}, 65); // 设置稍长于心跳的超时 console.log(用户 ${userId} 已连接连接ID: ${connectionId}); // 4. 监听客户端消息 ws.on(message, async (data) { try { const message JSON.parse(data); // 基础心跳包处理 if (message.type PING) { ws.send(JSON.stringify({ type: PONG, timestamp: Date.now() })); return; } // 业务消息转发到逻辑层消息队列 if (message.type CHAT) { await mqClient.publish(user_message_queue, JSON.stringify({ from: userId, to: message.to, content: message.content, msgId: uuidv4(), timestamp: Date.now() })); } } catch (err) { console.error(消息处理错误:, err); } }); // 5. 处理连接关闭 ws.on(close, async () { onlineClients.delete(userId); await redisClient.del(user:online:${userId}); console.log(用户 ${userId} 连接关闭); }); // 6. 发送连接成功确认 ws.send(JSON.stringify({ type: CONNECTED, connectionId })); });3.3 消息幂等与会话同步在分布式环境下消息重复推送和会话状态不一致是常见问题。消息幂等处理每条消息生成一个全局唯一的msgId如UUID。在接收端无论是用户还是客服处理消息前先用msgId去Redis查一下是否已处理过。如果已处理则直接丢弃或返回成功确保同一消息不会导致业务逻辑重复执行例如重复扣款、重复计数。// 在逻辑层消费消息时 const isProcessed await redisClient.setnx(msg_dedup:${msgId}, 1); if (isProcessed 0) { console.log(消息 ${msgId} 已处理跳过); return; // 幂等性检查防止重复消费 } await redisClient.expire(msg_dedup:${msgId}, 86400); // 24小时过期 // ... 继续处理消息业务逻辑会话状态同步客服可能在不同设备登录需要看到统一的会话列表和未读数。我们使用Redis的Hash结构存储会话最新消息和未读数。任何设备更新状态如已读时都通过逻辑层服务原子化地更新Redis中的状态并广播一个状态同步事件给该客服的所有在线连接触发各客户端本地状态更新。4. 性能优化实战4.1 连接池与资源管理虽然Node.js擅长高并发I/O但单个服务器能维护的WebSocket连接数仍受限于内存和文件描述符。我们做了以下优化调整系统限制增加服务器的最大文件打开数ulimit -n。心跳保活与空闲连接清理客户端每30秒发送一次心跳PING。服务器端设置45秒的超时检查如果超时未收到心跳则主动断开连接释放资源。这避免了大量“僵尸连接”占用资源。网关无状态化网关节点不存储会话数据只维护连接对象。用户连接可以连接到任意网关。用户在线状态和路由信息用户在哪个网关存储在Redis中逻辑层通过查询Redis就能将消息转发到正确的网关。4.2 基于Redis的智能分流算法当用户发起咨询时需要分配一个客服。简单的轮询分配可能造成客服负载不均。我们设计了一个简单的“智能”分流算法在Redis中为每个客服维护一个排序集合分数是当前接待的会话数。当用户需要分配客服时逻辑层从排序集合中取出分数最低即当前最闲的客服。分配后将该客服的分数加1。当客服结束一个会话时分数减1。这样就能动态地将新用户分配给当前最空闲的客服实现负载均衡。// 简化版智能分配客服 async function assignCustomerService(userId) { const csSetKey customer_service_load; // 获取负载最轻的3个客服避免第一名刚好不可用 const lightestCS await redisClient.zrange(csSetKey, 0, 2, WITHSCORES); for (const [csId, score] of lightestCS) { // 检查客服是否在线有另一个在线状态集合 if (await isCSOnline(csId)) { // 分配成功增加其负载分数 await redisClient.zincrby(csSetKey, 1, csId); // 建立用户-客服的会话映射 await createSession(userId, csId); return csId; } } return null; // 无可用客服 }4.3 消息压缩与批处理对于历史消息拉取、客服端同时接收多个用户消息等场景网络传输量可能很大。消息压缩对于文本消息在逻辑层推送给网关前如果消息体较大如超过1KB我们使用zlib进行GZIP压缩。网关推送给客户端时会携带压缩标识客户端解压后再渲染。这在移动网络下节省流量效果明显。消息批处理对于非实时性要求极高的通知类消息如“客服正在输入...”状态、会话列表更新通知逻辑层会做一个短暂的缓冲如100毫秒将同一连接需要发送的多个消息合并成一个数组批量下发减少WebSocket帧的数量提升效率。5. 避坑指南那些我们踩过的雷心跳包超时设置心跳间隔和超时时间需要仔细权衡。间隔太短如5秒会增加不必要的流量和服务器检查负担间隔太长如60秒则导致发现死连接的延迟过高资源不能及时释放。我们最终设置为客户端30秒发一次服务器端45秒未收到则断开。关键点服务器超时时间一定要大于客户端发送间隔并考虑网络延迟。断线重连的竞态条件客户端断线重连时可能会在极短时间内建立新连接。如果旧连接的清理从onlineClientsMap和Redis中删除稍微慢了一点新连接就已经建立并写入了Redis。这时逻辑层根据Redis路由消息可能会错误地发到那个即将关闭的旧连接上。我们的解决方案是在Redis存储连接信息时不仅存gateway节点还存connectionId。逻辑层推送消息前会请网关节点确认该connectionId的连接是否依然活跃如果不活跃则触发一次清理并重新查询路由。敏感信息过滤方案所有用户发送的消息在进入MQ之前必须在逻辑层进行同步的敏感词过滤。我们使用DFA算法构建敏感词树过滤速度极快O(n)。对于疑似敏感内容可以选择直接拦截、替换为***或者打标后进入人工审核队列。绝对不要依赖前端或仅在展示时过滤那样数据在传输和存储过程中已经泄露了。6. 测试数据优化前后对比我们对优化前后的系统进行了压测使用JMeter模拟1万用户同时在线消息发送频率为每秒500条。指标传统轮询方案WebSocketMQ优化方案提升比例系统吞吐量 (QPS)~1,200~4,800300%平均响应时间3500ms120ms降低96%网关服务器内存占用4.2 GB1.8 GB降低57%单机最大支持连接数~5,000~25,000400%消息送达延迟 (P99)2.8s105ms降低96%可以看到优化后的方案在吞吐量、延迟和资源消耗上都有数量级的提升。尤其是在响应时间和延迟上从秒级降到毫秒级用户体验的改善是颠覆性的。写在最后这次重构让我们深刻体会到对于高并发的实时交互场景正确的技术选型和架构设计是多么重要。从被动的轮询切换到主动的WebSocket推送加上消息队列的缓冲和解耦整个系统的健壮性和扩展性都得到了质的飞跃。当然系统还有可以继续优化的地方。比如我们目前的消息都是“在线投递”如果用户连接断开期间有消息会在其重连后拉取。但这只是一个简单的拉取没有做到完整的“离线消息”存储与可靠投递保障比如确保必达、按序投递。如果让你来设计一个支持海量用户的离线消息系统你会如何平衡存储成本、推送及时性和消息顺序的一致性呢这是一个值得深入思考的问题。希望这篇分享能对大家有所帮助。架构设计没有银弹最适合自己业务场景的就是最好的。