Tokio Select 超时与取消

示例说明

async 操作没有内建时限，永远等待会拖垮上层。select! 宏把多个 future 排成赛跑：工作分支与定时器分支同时推进，任一完成即取其结果并自动丢弃另一个——被丢弃的 future 在下一个 await 点优雅终止，资源随析构释放。这就是 Rust 异步的超时原语，比回调式超时干净得多。

技术要点

• select 分支完成即取消其余分支
• 被丢弃分支在 await 点安全终止
• timeout 组合子是 select 的封装

代码解读

with_deadline 里行情请求与 150 毫秒定时器竞争——真实请求要三百毫秒所以超时分支胜出，返回带截止时间的错误。stream_with_timeout 在 recv 循环上叠加超时：一百二十毫秒无消息即放弃通道，发送端的 send 错误让它感知退出。最后的 timeout 组合子演示三层超时的最外层封装。

使用建议

select 分支有随机优先级，同时完成时不保证顺序。被取消分支持有的锁会在 drop 时释放。生产超时值按下游 P99 设定并分层递减。

代码示例

use std::time::Duration;
use tokio::time::{sleep, timeout};

async fn fetch_quote(symbol: &'static str) -> String {
    sleep(Duration::from_millis(300)).await;
    format!("{symbol}=142.50")
}

async fn with_deadline(symbol: &'static str) -> Result<String, String> {
    let deadline = Duration::from_millis(150);

    // select 让慢操作与超时竞争，先完成者胜出
    tokio::select! {
        quote = fetch_quote(symbol) => Ok(quote),
        _ = sleep(deadline) => Err(format!("{symbol} exceeded {deadline:?}")),
    }
}

async fn stream_with_timeout() -> Vec<String> {
    let mut received = Vec::new();
    let (tx, mut rx) = tokio::sync::mpsc::channel(8);

    tokio::spawn(async move {
        for i in 0..5 {
            sleep(Duration::from_millis(40)).await;
            if tx.send(format!("tick-{i}")).await.is_err() {
                break; // 接收端已放弃
            }
        }
    });

    loop {
        let next = tokio::select! {
            msg = rx.recv() => msg,
            _ = sleep(Duration::from_millis(120)) => None,
        };
        match next {
            Some(msg) => received.push(msg),
            None => break, // 通道关闭或超时
        }
    }
    received
}

#[tokio::main]
async fn main() {
    println!("{:?}", with_deadline("APPL").await);

    let wrapped = timeout(Duration::from_millis(500), fetch_quote("MSFT")).await;
    println!("{:?}", wrapped.map(|s| s.len()));

    println!("{:?}", stream_with_timeout().await);
}