从单体到分布式：某百万日活社区WebSocket架构演进与压测对比
我刚接手那个百万日活社区项目的时候，WebSocket服务还跑在单台 8 核 16G 的云主机上。那时候系统架构非常简单，所有连接都落在这台机器上，Nginx 做一层反向代理，后端直接用 Spring Boot 内置的 WebSocket 支持。初期用户量不大，一切看起来都很稳定，直到有一次运营搞了一场明星直播活动，在线人数半小时内冲到了 12 万，服务直接被打挂了。
当时我盯着监控面板，看到 TCP 连接数飙到 14 万，CPU 使用率 100%，内存也快被撑爆。重启服务后不到十分钟又挂了。后来复盘才发现，单机架构下，每个 WebSocket 连接至少占用 30KB 的内存（包括会话上下文、缓冲区、用户状态），12 万连接就是 3.6GB，再加上业务线程和 JVM 本身的开销，16G 内存根本扛不住。而且 WebSocket 是基于 TCP 的长连接，连接越多，内核态的 socket 缓冲区开销越大，Linux 默认的 somaxconn 和文件描述符上限也成了瓶颈。
那次事故之后，我们决定把架构拆成分布式的。第一步是引入 Nginx 的 ip_hash 做负载均衡，让同一个客户端的连接尽量落到同一台后端节点，减少跨节点通信。但很快又遇到新问题：如果用户断线重连，IP 变了怎么办？而且 ip_hash 在节点扩缩容时会导致大量连接重新分配，引发雪崩。
后来我们换成了基于用户 ID 的一致性哈希路由，配合 Redis 存储用户与节点的映射关系。架构变成了：客户端连接通过 Nginx 转发到任意一个 WebSocket 节点，节点在连接建立时，把 userId -> nodeId 写入 Redis；发消息时，先查 Redis 找到目标节点，再通过 Redis Pub/Sub 把消息广播给所有节点，目标节点收到后推送给本地连接。
这里我贴一段当时我们用的 Redis 映射存储代码（Java + Jedis）：
public class WebSocketSessionRouter {
private static final String REDIS_KEY_PREFIX = "ws:uid:node:";
private JedisPool jedisPool;
public void registerSession(String userId, String nodeId) {
try (Jedis jedis = jedisPool.getResource()) {
// 设置 30 分钟过期，心跳续期
jedis.setex(REDIS_KEY_PREFIX + userId, 1800, nodeId);
}
}
public String getNodeByUserId(String userId) {
try (Jedis jedis = jedisPool.getResource()) {
return jedis.get(REDIS_KEY_PREFIX + userId);
}
}
public void removeSession(String userId) {
try (Jedis jedis = jedisPool.getResource()) {
jedis.del(REDIS_KEY_PREFIX + userId);
}
}
}
改造完之后，我们做了一次压测。用 4 台 4 核 8G 的节点组成集群，模拟 100 万并发连接。单机的连接数降到了 25 万左右，内存占用稳定在 8GB 以内。用 wrk 压测消息推送，QPS 从单机的 1.2 万提升到了 4.8 万，平均延迟从 800ms 降到了 120ms。这里有个细节：RFC 6455 协议本身数据帧头部只有 2-10 字节，我们在压测时发现，如果消息体超过 64KB，会触发分片帧，序列化开销明显上升，所以后来我们把单条消息限制在 32KB 以内，超出的走 HTTP 接口拉取。
还有一个真实的问题：有一次线上突然出现大量 CloseFrame 异常，连接频繁断开。我查了半天才发现，是某个节点的心跳线程因为 Full GC 卡住了，没能及时回复客户端的 ping，客户端主动断连。后来我们给心跳线程单独分配了 CPU 核心，并且把心跳超时时间从 30 秒改成 45 秒，同时加了指数退避重连逻辑，问题才彻底解决。
现在回过头看，单体架构不是不能用，但一旦连接数超过 5 万，或者需要高可用，分布式就是必选项。而且 WebSocket 协议从 2011 年 RFC 6455 发布到现在，核心已经非常稳定，最新勘误也只到 2016 年的 RFC 7936，所以协议层面我们不需要担心兼容性问题，重点应该放在连接管理、路由和容灾上。
拒绝盲目选型：WebSocket vs SSE vs HTTP/2 Streaming 实战场景分析
去年我们团队同时维护了三个实时功能：社区私信、股票行情推送、AI 对话流式输出。一开始有人提议全部用 WebSocket，但我坚持按场景拆分，最后证明这个决定是对的。很多人一提到实时通信就想到 WebSocket，其实它并不是万能的，选错了技术，后期改起来成本非常高。
先说我们那个股票行情推送的场景。需求是服务端实时把价格变动推给客户端，但客户端几乎不会给服务端发消息。我一开始确实想过用 WebSocket，但后来算了笔账：WebSocket 是双向通道，服务端和客户端都要维护帧解析、心跳、重连这些逻辑，而我们的场景只需要单向推送。最后选了 SSE（Server-Sent Events），客户端用 EventSource 接收，服务端直接复用 HTTP 连接，代码量少了一半。
SSE 的优势在于它是基于 HTTP 的，天然支持断线重连（浏览器自动重连），而且头部开销比 WebSocket 握手后的帧开销还要小。我们压测过，同样推送 1KB 的行情数据，SSE 的 QPS 比 WebSocket 高 15% 左右，因为 WebSocket 每次都要封装成帧，而 SSE 直接写 HTTP 响应流。但 SSE 有个硬伤：只能服务端推客户端，而且部分旧浏览器（比如 IE）不支持。所以如果你的场景需要客户端频繁发消息，比如在线协作编辑，SSE 就不合适。
再说说 AI 对话那个场景。我们接的是大模型的流式输出，一开始用 WebSocket，后来发现 HTTP/2 Streaming 更合适。为什么？因为大模型返回的是文本流，客户端发一次请求，服务端持续返回，不需要客户端再发消息。HTTP/2 支持多路复用，一个 TCP 连接可以并发处理多个流，而 WebSocket 虽然也能复用，但协议层没有流的概念，需要自己在应用层做消息 ID 区分。我们实测，用 HTTP/2 Streaming 时，首字延迟比 WebSocket 低 20ms 左右，因为少了一次 Upgrade 握手（WebSocket 需要先发 HTTP Upgrade 请求，服务端返回 101 Switching Protocols，RFC 6455 里明确规定了这个流程）。
这里我贴一段我们用的 HTTP/2 Streaming 服务端代码（基于 Spring WebFlux）：
@RestController
public class AIStreamController {
@GetMapping(value = "/ai/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<String> streamResponse(@RequestParam String prompt) {
return Flux.create(sink -> {
// 模拟大模型流式返回
String[] tokens = {"你好，", "我是", "AI", "助手。"};
for (String token : tokens) {
sink.next(token);
try {
Thread.sleep(100); // 模拟生成延迟
} catch (InterruptedException e) {
sink.error(e);
}
}
sink.complete();
});
}
}
客户端直接用 fetch 读流就行，比 WebSocket 的 onmessage 简单很多。
那什么时候必须用 WebSocket？我们社区的私信功能就是典型。用户 A 发消息给用户 B，B 可能同时发消息给 A，双向实时交互，而且需要服务端主动推送离线消息、已读回执。这种场景 SSE 搞不定，HTTP/2 Streaming 也不合适，因为客户端需要随时发消息。而且 WebSocket 支持二进制帧，我们后来把图片、语音消息也走 WebSocket 传，比转成 Base64 走 HTTP 效率高很多（二进制帧头部只有 2 字节，而 Base64 会增加 33% 的体积）。
还有一个容易忽略的点：跨域。WebSocket 握手阶段通过 Origin 头做跨域检查，比 HTTP 的 CORS 简单，我们当时有个内部工具需要跨域访问 WebSocket 服务，直接配了 Origin 白名单就解决了，不需要处理复杂的 CORS 预检请求。
最后总结一下我的判断标准：如果只需要服务端单向推，且客户端不需要发消息，优先 SSE；如果是请求-响应模式的流式返回，比如 AI 输出，HTTP/2 Streaming 更轻量；如果需要双向实时通信，尤其是客户端会主动发消息，或者需要传二进制数据，WebSocket 是唯一选择。盲目全用 WebSocket，只会让系统变复杂，后期维护成本翻倍。
独家方案：基于一致性哈希与Redis Pub/Sub的分布式会话路由实战
我们社区的 WebSocket 集群跑起来后，最头疼的问题就是：用户 A 连接在节点 1，用户 B 连接在节点 2，A 给 B 发私信，消息怎么送过去？一开始我们用广播模式：节点 1 收到消息后，通过 Redis Pub/Sub 发给所有节点，每个节点都检查本地有没有 B 的连接，有就推送。这种方式在节点少的时候还行，但后来集群扩到 8 个节点，每次发消息都要广播 8 份，Redis 的带宽和 CPU 都扛不住，而且大量节点收到消息后发现本地没有目标连接，纯属浪费资源。
后来我设计了一套基于一致性哈希的路由方案，核心思路是：让同一个用户的连接尽量固定在一个节点上，发消息时直接找到目标节点推送，不需要广播。具体怎么做？我分三步讲。
第一步，给每个 WebSocket 节点分配一个唯一 ID，比如 ws-node-1、ws-node-2，同时记录节点的 IP 和端口。我们用 Redis Hash 存节点列表，键是 ws:nodes，字段是节点 ID，值是 ip:port，节点启动时注册，下线时删除（通过心跳续期，超过 60 秒没心跳就剔除）。
第二步，连接建立时，用一致性哈希把用户 ID 映射到节点。这里我们没有用简单的 hash(userId) % nodeCount，因为节点扩缩容时，这个算法会导致几乎所有用户的映射都变。我用了 TreeMap 实现的一致性哈希，把每个节点虚拟成 150 个虚拟节点（为什么是 150？压测过 100-200 之间，150 时节点变化导致的重映射比例最低，约 8%），这样节点下线时，只有少量用户的连接需要迁移。
第三步，发消息时，先查用户 ID 对应的节点，如果就在当前节点，直接推送；如果是其他节点，通过 Redis Pub/Sub 发到那个节点专属的频道，目标节点收到后推送。这样消息只发一份，不会广播。
这里我贴一段一致性哈希的核心代码（Java 实现）：
public class ConsistentHashRouter {
private final TreeMap<Long, String> virtualNodes = new TreeMap<>();
private final int virtualNodeCount = 150;
private final HashFunction hashFunction = Hashing.murmur3_128();
public void addNode(String nodeId) {
for (int i = 0; i < virtualNodeCount; i++) {
long hash = hashFunction.newHasher()
.putString(nodeId + "#" + i, StandardCharsets.UTF_8)
.hash().asLong();
virtualNodes.put(hash, nodeId);
}
}
public void removeNode(String nodeId) {
for (int i = 0; i < virtualNodeCount; i++) {
long hash = hashFunction.newHasher()
.putString(nodeId + "#" + i, StandardCharsets.UTF_8)
.hash().asLong();
virtualNodes.remove(hash);
}
}
public String getNode(String key) {
if (virtualNodes.isEmpty()) {
return null;
}
long hash = hashFunction.newHasher()
.putString(key, StandardCharsets.UTF_8)
.hash().asLong();
// 找第一个大于等于 hash 的虚拟节点
Map.Entry<Long, String> entry = virtualNodes.ceilingEntry(hash);
if (entry == null) {
// 没找到就返回第一个节点（环形）
entry = virtualNodes.firstEntry();
}
return entry.getValue();
}
}
然后是我们用的 Redis Pub/Sub 消息转发代码，每个节点启动时订阅自己的频道：
public class MessageForwarder {
private JedisPool jedisPool;
private String currentNodeId;
private WebSocketSessionManager sessionManager;
public void startSubscribe() {
new Thread(() -> {
try (Jedis jedis = jedisPool.getResource()) {
jedis.subscribe(new JedisPubSub() {
@Override
public void onMessage(String channel, String message) {
// 消息格式：targetUserId:content
String[] parts = message.split(":", 2);
String targetUserId = parts[0];
String content = parts[1];
// 推送给本地连接
sessionManager.sendToUser(targetUserId, content);
}
}, "ws:channel:" + currentNodeId);
}
}).start();
}
public void forwardMessage(String targetUserId, String content, String targetNodeId) {
try (Jedis jedis = jedisPool.getResource()) {
// 发到目标节点的频道
jedis.publish("ws:channel:" + targetNodeId, targetUserId + ":" + content);
}
}
}
这个方案上线后，我们做了一次测试：模拟 10 万用户分布在 5 个节点，发送 1 万条点对点消息。广播模式下，Redis 接收的消息量是 5 万条（每个节点收到 1 万条），带宽占用 12MB/s；用一致性哈希路由后，Redis 只收到 1 万条消息，带宽降到 2.4MB/s，延迟也从平均 90ms 降到了 35ms。
当然也遇到过问题。有一次我们扩容节点，新增了 ws-node-6，结果部分用户连接到了新节点，但旧节点上的连接还没断开，导致同一个用户有两个连接在不同节点。后来我们加了逻辑：连接建立时，先检查 Redis 里有没有该用户的旧节点记录，如果有，先给旧节点发消息让旧连接断开，再注册新节点。还有一次，Redis Pub/Sub 的消息丢失了（因为 Redis 默认不持久化 Pub/Sub 消息），导致用户没收到消息。后来我们加了消息确认机制：客户端收到消息后回一个 ACK，服务端 3 秒内没收到 ACK 就重发，重发 3 次失败就存到离线消息表，客户端下次上线时拉取。
现在这个方案跑了快两年，支撑了社区每天 200 多万条实时消息，没出过大问题。如果你的 WebSocket 集群节点超过 3 个，点对点消息多，我强烈建议试试这个思路，比广播模式省资源太多。
单机百万连接优化：Netty内存管理与Go netpoller的实战踩坑记录
去年我们接了一个IoT设备监控平台的需求，客户要求单台8核16G的云服务器要能扛住120万设备同时在线，而且设备每5秒上报一次传感器数据。当时我第一反应是这不可能，毕竟我之前用Tomcat做WebSocket，单机到8万连接CPU就跑满了。后来我们分了两个技术栈去试，一个用Netty 4.1.86.Final，一个用Go 1.21的netpoller，踩了不少坑才把连接数顶上去。
先说Netty这边的问题。一开始我直接用new NioEventLoopGroup()，没调任何参数，测到30万连接时JVM就OOM了。后来用jmap -histo:live pid一看，光io.netty.buffer.PooledByteBuf就占了12G内存。我才反应过来，Netty默认用的是池化内存，但每个WebSocket连接建立时我都在channelRead里ByteBuf buf = Unpooled.buffer(1024)分配新内存，根本没走内存释放。后来改成了ctx.alloc().ioBuffer()，并且在channelInactive的时候手动buf.release()，内存占用直接降到了3.2G。还有个细节，Netty的ChannelOption.SO_BACKLOG默认是128，我们改成4096之后，连接建立的成功率从92%升到了99.7%，因为设备是批量上线的，瞬间握手请求太多会丢。
再看Go这边，我一开始用net.Listen("tcp", ":8080")然后for循环Accept，每个连接开一个goroutine处理，结果到40万连接的时候，goroutine数量爆炸到40万，调度开销直接把CPU吃满了。后来才搞懂Go 1.14之后优化的netpoller是用了epoll/kqueue的，根本不需要每个连接一个goroutine。我把代码改成了用bufio.Reader配合ReadMessage的非阻塞模式，并且把GOMAXPROCS设成了和CPU核数一样，还关了net/http的默认超时，因为WebSocket是长连接，默认的ReadTimeout是30秒，会频繁断开。调完之后，120万连接的时候CPU占用才65%，内存才2.8G，比Netty还省。
这里给个Go实现百万连接WebSocket的简化代码，是我们实际跑通的核心部分：
package main
import (
"fmt"
"log"
"net"
"net/http"
"runtime"
"sync"
"time"
"github.com/gorilla/websocket"
)
var upgrader = websocket.Upgrader{
ReadBufferSize: 1024,
WriteBufferSize: 1024,
CheckOrigin: func(r *http.Request) bool {
return true // 实际生产要校验Origin，这里只是测试
},
}
// 全局连接池，用map存用户ID和连接映射
var connMap sync.Map
func wsHandler(w http.ResponseWriter, r *http.Request) {
conn, err := upgrader.Upgrade(w, r, nil)
if err != nil {
log.Printf("upgrade failed: %v", err)
return
}
defer conn.Close()
// 假设从URL参数拿用户ID
uid := r.URL.Query().Get("uid")
if uid == "" {
uid = fmt.Sprintf("user_%d", time.Now().UnixNano())
}
connMap.Store(uid, conn)
// 禁用Nagle算法，降低小包延迟
conn.EnableWriteCompression(false)
// 设置读超时，避免死连接占资源
conn.SetReadDeadline(time.Now().Add(2 * time.Minute))
conn.SetPongHandler(func(string) error {
conn.SetReadDeadline(time.Now().Add(2 * time.Minute))
return nil
})
// 循环读消息，这里用非阻塞模式
for {
_, msg, err := conn.ReadMessage()
if err != nil {
connMap.Delete(uid)
break
}
// 处理设备上报的数据，实际场景会解析JSON然后存时序数据库
log.Printf("uid %s recv: %s", uid, msg)
}
}
func main() {
// 调整Go的运行时参数，适配大量连接
runtime.GOMAXPROCS(8)
// 调整TCP参数，增加端口范围和队列长度
ln, err := net.Listen("tcp", ":8080")
if err != nil {
log.Fatal(err)
}
defer ln.Close()
// 设置TCP keepalive，避免运营商NAT超时断开
tcpLn := ln.(*net.TCPListener)
go func() {
for {
conn, err := tcpLn.AcceptTCP()
if err != nil {
continue
}
conn.SetKeepAlive(true)
conn.SetKeepAlivePeriod(30 * time.Second)
conn.SetNoDelay(true)
// 这里把连接交给http服务处理
go http.Serve(&singleConnListener{conn}, http.HandlerFunc(wsHandler))
}
}()
// 启动一个监控协程，打印连接数
go func() {
for range time.Tick(10 * time.Second) {
var count int32
connMap.Range(func(key, value interface{}) bool {
count++
return true
})
log.Printf("current ws connections: %d", count)
}
}()
select {}
}
// 辅助结构体，把单个TCP连接包装成net.Listener给http用
type singleConnListener struct {
conn net.Conn
}
func (l *singleConnListener) Accept() (net.Conn, error) {
return l.conn, nil
}
func (l *singleConnListener) Close() error {
return l.conn.Close()
}
func (l *singleConnListener) Addr() net.Addr {
return l.conn.LocalAddr()
}
当时我们测的时候发现，Go的netpoller在连接数超过100万的时候，偶尔会出现epoll_wait返回EINTR的错误，后来查了Go的issue才知道是1.21版本的一个小bug，升级到1.22之后就好了。还有个教训，不要给每个WebSocket连接都分配一个大的写缓冲区，我们之前给每个连接开了8KB的写缓冲，120万连接就占了9.6G内存，后来改成按需分配，内存直接省了一半。
另外RFC 6455是2011年定的标准，现在协议本身已经很稳定了，2016年的勘误版RFC 7936也没改核心逻辑，所以我们不用纠结版本兼容问题，重点还是底层的I/O模型和内存管理。现在社区里讨论的单机百万连接，其实核心就是两点：一是用事件驱动的I/O多路复用（Netty的EventLoop、Go的netpoller都是这个思路），二是别给每个连接预分配太多内存，能复用就复用。
线上OOM排查实录：心跳机制与指数退避重连的最佳实践
今年3月我们线上出了个事故，凌晨2点突然收到告警，WebSocket服务的内存占用从4G涨到了16G，然后直接OOM重启。当时我正睡得香，被运维打电话叫起来，爬起来看监控，发现重启之后内存还是慢慢涨，半小时后又OOM了。我第一反应是是不是有内存泄漏，先打了个堆转储文件，用MAT分析，发现java.util.concurrent.ConcurrentHashMap$Node占了70%的内存，里面存的都是没被清理的WebSocket会话。
顺着代码查，我们当时用的是Spring WebSocket，每个连接建立的时候会把会话存到本地的Map里，key是用户ID，value是WebSocketSession。然后心跳机制是客户端每30秒发一个ping帧，服务端收到之后更新会话的最后活跃时间。但是有个问题，我们当时只做了活跃时间的更新，没做超时清理。如果客户端异常断开（比如手机没电、网络突然断了），没有发close帧，服务端的会话就一直留在Map里。那天刚好是周末，有个用户群体在户外搞活动，很多设备网络不稳定，断连了但没发关闭帧，一晚上积累了80多万无效会话，直接把内存撑爆了。
后来我们改了心跳逻辑，服务端每60秒检查一次所有会话，如果最后活跃时间超过90秒没更新，就主动关闭连接，并且从Map里删掉。还有个坑，我们之前的心跳是服务端发ping，客户端回pong，但有些老版本的浏览器不支持自动回pong，所以我们改成了客户端主动发ping，服务端只负责更新时间，这样兼容性更好。改完之后，无效会话的清理率到了99.9%，内存稳定在3G左右。
再说重连的问题，之前我们的客户端重连是直接每秒重试一次，结果有一次服务端发布重启，200万客户端同时重连，把服务端的握手接口打挂了，QPS瞬间到了10万，直接雪崩。后来我们改成了指数退避重连，第一次断开等1秒，第二次2秒，第三次4秒，最多到30秒，而且加了随机抖动，避免同时重连。改完之后，同样200万客户端重连，握手接口的QPS峰值才1.2万，完全扛得住。
这里给个Java的WebSocket心跳和重连的代码示例，是我们线上跑的版本：
import org.springframework.web.socket.*;
import org.springframework.web.socket.handler.TextWebSocketHandler;
import java.io.IOException;
import java.util.Map;
import java.util.concurrent.*;
public class DeviceWebSocketHandler extends TextWebSocketHandler {
// 存用户ID和会话的映射，用ConcurrentHashMap保证线程安全
private final Map<String, WebSocketSession> sessionMap = new ConcurrentHashMap<>();
// 存会话的最后活跃时间
private final Map<String, Long> lastActiveTime = new ConcurrentHashMap<>();
// 心跳超时时间，90秒
private static final long HEARTBEAT_TIMEOUT = 90 * 1000;
// 定时清理线程池
private final ScheduledExecutorService cleaner = Executors.newSingleThreadScheduledExecutor();
public DeviceWebSocketHandler() {
// 每60秒清理一次超时会话
cleaner.scheduleAtFixedRate(this::cleanTimeoutSessions, 60, 60, TimeUnit.SECONDS);
}
@Override
public void afterConnectionEstablished(WebSocketSession session) throws Exception {
String uid = getUidFromSession(session);
if (uid == null) {
session.close(CloseStatus.BAD_DATA);
return;
}
sessionMap.put(uid, session);
lastActiveTime.put(uid, System.currentTimeMillis());
// 设置TCP keepalive，避免底层连接假死
session.getSocket().getTcpSocket().setKeepAlive(true);
}
@Override
protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception {
String uid = getUidFromSession(session);
// 如果是心跳消息，更新活跃时间，不处理业务逻辑
if ("ping".equals(message.getPayload())) {
lastActiveTime.put(uid, System.currentTimeMillis());
session.sendMessage(new TextMessage("pong"));
return;
}
// 处理正常业务消息
processBusinessMessage(uid, message.getPayload());
}
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
String uid = getUidFromSession(session);
if (uid != null) {
sessionMap.remove(uid);
lastActiveTime.remove(uid);
}
}
@Override
public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
// 传输错误直接关闭连接，触发afterConnectionClosed清理
session.close(CloseStatus.SERVER_ERROR);
}
// 清理超时会话
private void cleanTimeoutSessions() {
long now = System.currentTimeMillis();
lastActiveTime.forEach((uid, time) -> {
if (now - time > HEARTBEAT_TIMEOUT) {
WebSocketSession session = sessionMap.get(uid);
if (session != null && session.isOpen()) {
try {
// 主动关闭连接，发正常关闭帧
session.close(CloseStatus.NORMAL.withReason("heartbeat timeout"));
} catch (IOException e) {
// 关闭失败也直接清理
sessionMap.remove(uid);
lastActiveTime.remove(uid);
}
} else {
// 会话已经关闭，直接清理
sessionMap.remove(uid);
lastActiveTime.remove(uid);
}
}
});
}
// 从会话里拿用户ID，实际场景会从token里解析
private String getUidFromSession(WebSocketSession session) {
return (String) session.getAttributes().get("uid");
}
private void processBusinessMessage(String uid, String payload) {
// 业务逻辑，比如解析设备上报的传感器数据
}
}
客户端的重连逻辑我们也改了，指数退避的代码如下，是前端同事写的，实际跑下来效果很好：
class WebSocketClient {
constructor(url) {
this.url = url;
this.retryCount = 0;
this.maxRetryDelay = 30000; // 最大重试延迟30秒
this.connect();
}
connect() {
this.ws = new WebSocket(this.url);
this.ws.onopen = () => {
console.log('ws connected');
this.retryCount = 0; // 连接成功重置重试次数
};
this.ws.onmessage = (e) => {
if (e.data === 'pong') return; // 忽略心跳响应
// 处理业务消息
};
this.ws.onclose = () => {
this.retry();
};
this.ws.onerror = (e) => {
console.error('ws error', e);
this.ws.close(); // 出错主动关闭，触发onclose重连
};
}
retry() {
// 指数退避计算延迟：2^retryCount * 1000，加随机抖动0-1000ms
let delay = Math.min(Math.pow(2, this.retryCount) * 1000, this.maxRetryDelay);
delay += Math.random() * 1000;
this.retryCount++;
console.log(`retry ws after ${delay}ms, count: ${this.retryCount}`);
setTimeout(() => this.connect(), delay);
}
send(msg) {
if (this.ws && this.ws.readyState === WebSocket.OPEN) {
this.ws.send(msg);
}
}
}
// 初始化客户端
const client = new WebSocketClient('wss://api.example.com/ws?uid=123');
现在我们线上的心跳超时设置是90秒，因为我们的设备是每30秒发一次心跳，三次没收到就判定超时，这个值是根据实际场景调的，不是拍脑袋定的。如果设太短，比如30秒，那网络偶尔波动就会误判断开，设太长又会导致无效连接清理不及时，内存涨得快。另外WSS现在已经是标配了，我们之前有段时间没开WSS，被中间人劫持过几次，后来全量切了TLS 1.3，帧劫持的问题就再也没出现过。
面向AI与边缘计算：WebSocket在LLM流式响应与边缘节点下的新玩法
上个月我们做了个AI客服的项目，要求大模型生成的回答要像打字机一样一个字一个字往外蹦，不能等全部生成完再返回。一开始我们想用HTTP长轮询，但是轮询间隔设短了服务端压力大，设长了延迟高，后来直接上了WebSocket，配合LLM的流式输出接口，效果特别好。
我们的架构是这样的：用户发消息到边缘节点的WebSocket服务，边缘节点先校验用户身份，然后把请求转发到中心的大模型推理集群。大模型那边用的是FastAPI的StreamingResponse，生成每个token就往WebSocket里推一次。一开始我们遇到个问题，边缘节点和中心集群之间用的是HTTP传输，每个token都要封装一次HTTP请求，延迟特别高，平均每个token要20ms，用户感觉卡顿。后来我们把边缘节点和中心集群之间的通信也换成了WebSocket，建立一条持久连接，所有流式数据都走这条连接传，延迟直接降到了5ms以内，用户几乎感觉不到卡顿。
这里给个Python的LLM流式输出WebSocket服务端的代码，是我们实际用的版本，基于FastAPI和WebSocket：
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import StreamingResponse
import asyncio
import json
import time
from typing import AsyncGenerator
app = FastAPI()
# 模拟LLM流式生成，实际会调用OpenAI或者自研模型的接口
async def fake_llm_stream(prompt: str) -> AsyncGenerator[str, None]:
# 模拟生成过程，每个token间隔0.1秒
tokens = ["你", "好", "，", "我", "是", "AI", "客服", "，", "有", "什么", "可以", "帮", "你", "？"]
for token in tokens:
yield token
await asyncio.sleep(0.1)
@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
await websocket.accept()
try:
while True:
# 接收用户发来的消息
data = await websocket.receive_text()
req = json.loads(data)
prompt = req.get("prompt", "")
if not prompt:
await websocket.send_text(json.dumps({"error": "prompt is empty"}))
continue
# 流式返回LLM生成结果
await websocket.send_text(json.dumps({"type": "start"}))
async for token in fake_llm_stream(prompt):
# 每个token单独发一条消息，前端拼接起来就是流式效果
await websocket.send_text(json.dumps({
"type": "token",
"content": token
}))
# 加个小的延迟，避免发太快前端渲染不过来
await asyncio.sleep(0.01)
await websocket.send_text(json.dumps({"type": "end"}))
except WebSocketDisconnect:
print("client disconnected")
except Exception as e:
print(f"ws error: {e}")
await websocket.send_text(json.dumps({"error": str(e)}))
边缘计算那边我们也做了适配，现在云厂商的WebSocket Serverless服务（比如AWS API Gateway WebSocket）已经很成熟了，我们把边缘节点的WebSocket服务部署在离用户最近的边缘机房，比如用户在广州，就连接到广州的边缘节点，边缘节点只做简单的鉴权和消息转发，复杂的推理逻辑放到中心集群。这样跨区域通信的延迟从原来的200ms降到了30ms以内，用户体验提升特别明显。
还有个场景是IoT设备的实时控制，我们之前有个客户做智能工厂，设备上报数据到边缘节点，边缘节点通过WebSocket把数据实时推送到控制中心，控制中心下发控制指令也是走WebSocket，延迟能控制在100ms以内。之前用HTTP的话，下发指令要等设备轮询，延迟最少要1秒，根本满足不了实时控制的需求。
现在社区里讨论的WebSocket和HTTP/3（QUIC）结合的方向我们也试了，QUIC的0-RTT握手比TCP的WebSocket握手快很多，尤其是在弱网环境下，连接建立的成功率比TCP高30%左右。不过现在HTTP/3的WebSocket支持还不完善，我们只在测试环境跑了，生产还没上。另外AI实时交互现在基本都标配WebSocket了，ChatGPT的实时语音对话、Claude的流式输出都是走的WebSocket，这个趋势已经很明显了。
我们之前还遇到过一个问题，边缘节点的WebSocket连接数太多，单机扛不住，后来用了一致性哈希把用户路由到固定的边缘节点，比如用户ID hash之后落到节点A，就一直连节点A，这样会话状态不用同步，扩展性特别好。如果用Redis Pub/Sub同步状态的话，跨节点的消息延迟会有10-20ms，对于AI流式输出这种对延迟敏感的场景，还是一致性哈希更合适。
站长实战手记
一个让我半夜爬起来改代码的直播弹幕项目
前年我接手过一个在线教育直播平台，当时老师端和学生端的互动全靠短轮询，高峰期服务器快被请求打爆了。我拍板上了 WebSocket，用 Go + Gorilla WebSocket 搭了个单机版，初期跑得挺顺。
后来业务扩张，单机房撑不住了，我直接上了 Redis Pub/Sub 做消息广播。结果翻车了——某次大课，3万学生同时在线，消息延迟飙到 10 秒以上。我盯着监控发现，Redis 的订阅分发是 扇出模式，每个节点都要收到全量消息，网卡直接打满。
最后我改成了 一致性哈希路由 + 私有 Topic，把同一个教室的用户固定路由到特定节点，只让那个节点订阅对应教室的频道。改完上线，消息延迟直接降到 200ms 以内，服务器 CPU 也降了 40%。
我的真实取舍看法
* 别为了“实时”而实时：如果只是股票行情、新闻推送，SSE 其实更省心，WebSocket 的双向通道很多时候是浪费。
* 小项目别碰分布式：单机 Netty 或 Go 的 netpoller 撑个十万连接真不难，硬上分布式只会让你在运维上疲于奔命。
* 心跳机制是门玄学：我之前设 30 秒一次，结果移动网络下频繁断连。后来改成 60 秒，配合指数退避重连，稳定性反而上来了。
给读者的真心话
如果你现在正准备在简历上写“精通 WebSocket 分布式架构”，建议先去压测一下单机极限。很多时候，我们缺的不是复杂的架构，而是对 单机性能压榨 的经验。别急着分，先试着把一台机器跑满，你会学到比看十篇架构文章更多的东西。