WebSocket连接异常中断时web::Payload::next()挂起问题求助

问题原因 #

当客户端未发送WebSocket Close帧就异常断开连接（如进程崩溃、网络中断），TCP连接可能进入半开状态：服务器端的读取操作不会立即返回错误或None，而是会一直阻塞等待数据，导致stream.next().await无限挂起。

解决方案 #

可以通过超时机制或心跳检测来主动识别异常连接，避免任务挂起。以下是具体实现方案：

方案1：给读取操作添加超时 #

使用tokio::time::timeout为每个stream.next()调用设置超时，超过指定时间无数据则判定连接异常断开：

use tokio::time::{timeout, Duration};
use actix_web::{web, Error as ActixError};
use bytes::Bytes;

async fn handle_frames(mut stream: web::Payload) -> Result<(), ActixError> {
    // 设置30秒读取超时，可根据业务调整
    const READ_TIMEOUT: Duration = Duration::from_secs(30);

    loop {
        match timeout(READ_TIMEOUT, stream.next()).await {
            Ok(Some(Ok(chunk))) => {
                // 处理收到的WebSocket数据块
                // process_chunk(chunk).await?;
            }
            Ok(Some(Err(e))) => {
                // 读取到明确错误（如连接正常关闭）
                log::error!("WebSocket读取错误: {}", e);
                break;
            }
            Ok(None) => {
                // 数据流正常结束
                break;
            }
            Err(_) => {
                // 超时触发，判定为连接异常断开
                log::info!("WebSocket读取超时，连接已断开");
                break;
            }
        }
    }
    Ok(())
}

方案2：给整个数据流包装超时 #

利用futures_util::StreamExt的timeout扩展方法，给整个Payload流添加全局超时：

use futures_util::StreamExt;
use tokio::time::Duration;
use actix_web::{web, Error as ActixError};
use bytes::Bytes;

async fn handle_frames(mut stream: web::Payload) -> Result<(), ActixError> {
    const READ_TIMEOUT: Duration = Duration::from_secs(30);
    // 给数据流添加超时包装
    let mut stream = stream.timeout(READ_TIMEOUT);

    while let Some(result) = stream.next().await {
        match result {
            Ok(Ok(chunk)) => {
                // 处理数据块
                // process_chunk(chunk).await?;
            }
            Ok(Err(e)) => {
                log::error!("WebSocket读取失败: {}", e);
                break;
            }
            Err(_) => {
                log::info!("读取超时，连接异常断开");
                break;
            }
        }
    }
    Ok(())
}

方案3：结合心跳机制（推荐） #

遵循WebSocket协议规范，定期向客户端发送PING帧，若未收到PONG响应则主动断开连接，同时配合超时机制：

use tokio::time::{interval, Duration};
use futures_util::{future::select, FutureExt, StreamExt};
use actix_web::{web, Error as ActixError};
use bytes::Bytes;
use actix_web::ws::{Frame, Message};
use tokio::sync::mpsc;

async fn handle_frames(
    mut stream: web::Payload,
    tx: mpsc::Sender<Result<Bytes, ActixError>>
) -> Result<(), ActixError> {
    // 心跳间隔10秒，读取超时30秒
    const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(10);
    const READ_TIMEOUT: Duration = Duration::from_secs(30);

    let mut heartbeat_timer = interval(HEARTBEAT_INTERVAL);
    let mut stream = stream.timeout(READ_TIMEOUT);

    loop {
        let result = select(
            stream.next().boxed(),
            heartbeat_timer.tick().boxed()
        ).await;

        match result {
            // 收到客户端数据
            futures_util::future::Either::Left((Some(frame_result), _)) => {
                match frame_result {
                    Ok(Ok(chunk)) => {
                        // 解析帧，若收到PONG则重置心跳（可选）
                        if let Ok(Message::Pong(_)) = Frame::parse(&chunk) {
                            heartbeat_timer.reset();
                        }
                        // 处理业务数据
                        // process_data(chunk).await?;
                    }
                    Ok(Err(e)) => {
                        log::error!("读取错误: {}", e);
                        break;
                    }
                    Err(_) => {
                        log::info!("读取超时，连接断开");
                        break;
                    }
                }
            }
            // 触发心跳，发送PING帧
            futures_util::future::Either::Right((_, _)) => {
                let ping_frame = Frame::Ping(Bytes::from_static(b""));
                if tx.send(Ok(ping_frame.into())).await.is_err() {
                    log::info!("发送PING失败，连接已断开");
                    break;
                }
            }
            // 数据流结束
            futures_util::future::Either::Left((None, _)) => {
                break;
            }
        }
    }

    // 主动发送Close帧清理连接
    let close_frame = Frame::Close(None);
    let _ = tx.send(Ok(close_frame.into())).await;
    Ok(())
}

总结 #

• 超时机制是快速解决挂起问题的基础方案，适合简单场景
• 心跳机制更符合WebSocket协议，能主动验证连接可用性，避免误判
• 两者结合使用可兼顾可靠性与及时性，是生产环境的推荐方案

内容的提问来源于stack exchange，提问作者ababo