Tokio Tungstenite WebSocket 双半拆分

示例说明

单个流对象上先读后写会把两个方向耦合成串行——发送队列被慢接收拖住。split 拆出所有权分离的两半：读半与写半各自被独立 future 驱动，select! 让两路事件公平竞争。协议层的 Ping 帧自动回 Pong 维持链路，应用层心跳用文本消息表达业务级探活，两层心跳各有分工。

技术要点

• split 让读写方向所有权独立
• 协议层 Pong 与应用层心跳并存
• select 驱动双向事件公平竞争

代码解读

连接建立后立刻拆分；select 的读分支处理文本消息、Ping 帧、错误与关闭四种情况；心跳分支每二十五秒发送一条 JSON 探活消息。收满五条消息后主动发送 Close 帧优雅收尾——比直接断链让对端能干净退出。

使用建议

重连逻辑应包装在连接管理器里维护退避状态。消息类型按协议选择二进制或文本。慢消费场景给写半加有界通道做缓冲。

代码示例

use futures_util::{SinkExt, StreamExt};
use tokio_tungstenite::{connect_async, tungstenite::Message};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let (ws_stream, _response) = connect_async("wss://stream.example.com/feed").await?;

    // 拆分为独立的读半与写半：两个方向并发推进互不阻塞
    let (mut write, mut read) = ws_stream.split();

    let mut heartbeat = tokio::time::interval(std::time::Duration::from_secs(25));
    let mut received = 0usize;

    loop {
        tokio::select! {
            // 读方向：服务器推送的消息
            msg = read.next() => {
                match msg {
                    Some(Ok(Message::Text(text))) => {
                        received += 1;
                        println!("recv: {} bytes", text.len());
                    }
                    Some(Ok(Message::Ping(ping))) => {
                        // 协议层心跳：自动回 pong
                        write.send(Message::Pong(ping)).await?;
                    }
                    Some(Ok(_)) => {}
                    Some(Err(err)) => return Err(err.into()),
                    None => break, // 服务器关闭连接
                }
            }

            // 写方向：应用层心跳与订阅指令
            _ = heartbeat.tick() => {
                write.send(Message::Text(r#"{"op":"ping"}"#.into())).await?;
            }
        }

        if received >= 5 {
            break;
        }
    }

    write.send(Message::Close(None)).await?;
    println!("received {received} messages");
    Ok(())
}