Rust WebSocket 长连接：真正难的不是收发消息

WebSocket 最容易让人误判的地方，是它看起来像“HTTP 升级一下，然后循环收发 JSON”。

demo 代码通常很短：连接升级，split 成读写两半，读到消息就广播。浏览器里打开两个页面，互相发消息，一切正常。

但生产长连接要面对的问题完全不同：

客户端静默断开
慢消费者拖垮广播
心跳谁发谁收
断线重连会不会重复登录
消息要不要缓冲
服务端关闭时怎么通知客户端

WebSocket 是一条长期占用资源的会话，不是一个不会结束的 HTTP 请求。

这篇不讲加密聊天里的端到端协议，也不讲 PVP 游戏里的房间同步。这里只讲通用 WebSocket 服务端：连接生命周期、心跳、写队列、慢消费者和关闭。

本文代码环境：

# Cargo.toml
[dependencies]
axum = { version = "0.8", features = ["ws"] }
tokio = { version = "1", features = ["full"] }
tracing = "0.1"

连接不是用户，连接只是会话

WebSocket handler 大概长这样：

use axum::{extract::ws::WebSocketUpgrade, response::IntoResponse};
 
async fn ws_handler(ws: WebSocketUpgrade) -> impl IntoResponse {
    ws.on_upgrade(handle_socket)
}

升级后拿到 socket：

use axum::extract::ws::{Message, WebSocket};
 
async fn handle_socket(mut socket: WebSocket) {
    while let Some(Ok(msg)) = socket.recv().await {
        if let Message::Text(text) = msg {
            tracing::info!(%text, "received message");
        }
    }
}

这只能处理最简单的读消息。

真实服务里，连接要有自己的身份和状态：

#[derive(Clone, Debug)]
struct Session {
    user_id: String,
    conn_id: String,
}

不要把“用户在线”简化成“有一个 WebSocket 连接”。

同一个用户可能多端在线；浏览器刷新会产生新连接；旧连接可能已经断了但服务端还没发现。用户、设备、连接，这三个概念最好分开。

读写要拆开，但状态不能到处共享

常见结构是把 socket 拆成读任务和写任务：

use axum::extract::ws::WebSocket;
use futures_util::{SinkExt, StreamExt};
use tokio::sync::mpsc;
 
async fn run_connection(socket: WebSocket) {
    let (mut writer, mut reader) = socket.split();
    let (tx, mut rx) = mpsc::channel::<Message>(64);
}

写任务只负责把队列里的消息发出去：

tokio::spawn(async move {
    while let Some(msg) = rx.recv().await {
        if writer.send(msg).await.is_err() {
            break;
        }
    }
});

读循环只负责把客户端消息翻译成命令：

while let Some(Ok(msg)) = reader.next().await {
    if let Message::Text(text) = msg {
        handle_client_text(text.to_string()).await;
    }
}

补一个业务入口：

async fn handle_client_text(text: String) {
    tracing::info!(%text, "client command");
}

这个边界很重要：读任务不直接改全局状态，写任务不理解业务命令。连接层只做协议翻译。

写队列必须有上限

给每个连接一个写队列很自然，但队列不能无限大。

use tokio::sync::mpsc;
 
type Outbox = mpsc::Sender<Message>;
 
fn new_outbox() -> (Outbox, mpsc::Receiver<Message>) {
    mpsc::channel(64)
}

如果客户端网络很慢，服务端发送速度会下降。没有上限的写队列会把广播消息全堆在内存里。

队列满了以后，要有明确策略：

场景	策略
聊天消息	可以短暂等待，超时断开
在线状态	队列满了直接丢旧状态
游戏快照	只保留最新快照
系统通知	失败后落库或等待客户端拉取

WebSocket 的慢消费者问题，本质还是背压。

一个客户端读得慢，不能拖垮整个房间或整个服务。

心跳不是装饰

TCP 半开连接很常见：客户端网络断了，服务端一段时间内并不知道。

WebSocket 要靠 ping/pong 或应用层心跳确认连接还活着。

一个简单的 ping 循环：

use std::time::Duration;
use axum::extract::ws::Message;
use tokio::time;
 
async fn heartbeat(tx: mpsc::Sender<Message>) {
    let mut tick = time::interval(Duration::from_secs(30));
    loop {
        tick.tick().await;
        if tx.send(Message::Ping(Vec::new().into())).await.is_err() {
            break;
        }
    }
}

只发 ping 还不够，还要记录最近一次收到客户端消息或 pong 的时间。超过阈值，就主动断开。

use std::time::{Duration, Instant};
 
fn is_idle_too_long(last_seen: Instant) -> bool {
    last_seen.elapsed() > Duration::from_secs(90)
}

心跳间隔不要太激进。移动网络、后台标签页、省电模式都会影响心跳稳定性。

我的偏好是：心跳用于发现死连接，不用于追求毫秒级在线状态。

广播要绕开慢连接

广播最容易写成这样：

async fn broadcast(clients: &[Outbox], msg: Message) {
    for tx in clients {
        let _ = tx.send(msg.clone()).await;
    }
}

这段代码的问题是：一个慢连接就能拖慢整个广播循环。

更保守的做法是快速投递，满了就把这个连接标记为慢消费者：

fn try_broadcast(clients: &[Outbox], msg: Message) -> usize {
    let mut dropped = 0;
    for tx in clients {
        if tx.try_send(msg.clone()).is_err() {
            dropped += 1;
        }
    }
    dropped
}

这不是说所有消息都该丢。

关键是广播层不能无限等待单个连接。必须把“投递失败”交给更高层策略：断开、降级、保留最新、落库补偿。

关闭也要发消息

服务端准备关闭连接时，最好发送 close frame：

use axum::extract::ws::{CloseFrame, Message};
 
fn server_shutdown_message() -> Message {
    Message::Close(Some(CloseFrame {
        code: 1001,
        reason: "server shutdown".into(),
    }))
}

这不是礼貌问题，是客户端体验问题。

客户端收到 close frame 后可以立即重连、切换节点或展示状态。如果服务端直接 drop socket，客户端只能等底层连接错误或自己的心跳超时。

优雅关闭里讲过：进程退出前要通知后台任务收尾。WebSocket 也一样，连接退出前要尽量告诉对端为什么断。

结论

WebSocket 长连接的核心不是“怎么收发消息”，而是“怎么管理一条会活很久的会话”。

可以直接记这几条：

用户、设备、连接要分开建模
读写任务可以拆开，但状态边界要清楚
每个连接的写队列必须有上限
心跳用来发现死连接，不是装饰
广播不能被慢连接拖住
队列满了要有业务策略
服务端关闭时尽量发送 close frame

WebSocket 是会话系统，不是无限 HTTP。

如果只写收发循环，demo 会很顺；如果不处理心跳、背压、慢消费者和关闭，生产环境里长连接会把问题慢慢积累到最难排查的时候。