在异步 Rust 里同时等待多个数据源，最直接的写法是 tokio::select!。但它有一个不太显眼的语义：select! 只负责「谁先就绪就返回谁」，不负责「等多久就不等了」。当某个分支既不返回数据也不报错——比如一个长时间静默的 WebSocket、一个对端迟迟不发心跳的 TCP 连接、一个被上游限流卡住的 HTTP body——整个 select! 就会一直挂着。上层传下来的 deadline 形同虚设。
常见的补救办法是给整个 select! 外面套一个大的 timeout，但这会带来两个新问题：一是超时后你无法区分「是哪一个分支卡住了」，二是某个分支已经读到一半的数据会被整体丢弃。本文给出另一种更细粒度的做法：把 timeout 下沉到 select! 的每个分支上，让每个数据源各自带自己的截止时间，同时保留「共享总 deadline」的能力。文中的代码全部基于 tokio = "1.40"（features 至少包含 macros、time、rt-multi-thread、io-util、sync、net），可直接 cargo run。
先看清 select! 的语义边界
写错这一点的代码通常长这样：
use tokio::time::{sleep, Duration};
async fn read_a() -> Option<u8> {
sleep(Duration::from_secs(3600)).await;
Some(1)
}
async fn read_b() -> Option<u8> {
sleep(Duration::from_millis(50)).await;
Some(2)
}
#[tokio::main]
async fn main() {
// 期望 1 秒后放弃，实际上：read_b 50ms 就绪，select! 立即返回 Some(2)
tokio::select! {
v = read_a() => println!("a => {:?}", v),
v = read_b() => println!("b => {:?}", v),
}
}这段代码能跑，输出是 b => Some(2)。它掩盖了真正的问题：一旦把 read_b 也换成一个长期静默的源，select! 会永久 pending。select! 本身没有时间维度。
另一个高频误解是 select! 加 else 分支能当超时用。不能。else 只在所有分支都被 disabled（即每个分支的 if 前置条件都为 false，或被 biased; 之前的手动禁用逻辑关掉）时执行，与时间无关。所以「加个 else 兜底」不是超时方案。
还有一点要记住：select! 的循环轮询是公平但仍可能饿死的。默认情况下 tokio 会在每次循环开始时随机选择分支的检查顺序，因此理论上存在某个分支连续多次被跳过的情况。如果你有明确优先级（比如「控制帧优先于数据帧」），用 biased;，但要清楚它会牺牲公平性。
核心模式：把 timeout 塞进每个分支
改造思路很直接：select! 的每个分支表达式本身就是一个 future，那就让这个 future 是 timeout(dur, fut) 的返回值，而不是 fut 本身。这样每个分支在「数据就绪」和「超时」两条路径上都必然完成，select! 不会再无限等待。
use std::time::{Duration, Instant};
use tokio::time::{sleep, timeout, Elapsed};
/// 模拟一个长期静默的数据源：只在 far_future 之后才产出
async fn silent_source() -> Result<u8, &'static str> {
sleep(Duration::from_secs(3600)).await;
Ok(0)
}
/// 模拟一个正常数据源
async fn fast_source() -> Result<u8, &'static str> {
sleep(Duration::from_millis(80)).await;
Ok(7)
}
#[tokio::main]
async fn main() {
let started = Instant::now();
tokio::select! {
r = timeout(Duration::from_millis(500), silent_source()) => {
match r {
Ok(Ok(v)) => println!("silent => Ok({v})"),
Ok(Err(e)) => println!("silent => 源内错误: {e}"),
Err(Elapsed) => println!("silent => 超时，耗时 {:?}", started.elapsed()),
}
}
r = timeout(Duration::from_millis(500), fast_source()) => {
match r {
Ok(Ok(v)) => println!("fast => Ok({v})"),
Ok(Err(e)) => println!("fast => 源内错误: {e}"),
Err(Elapsed) => println!("fast => 超时，耗时 {:?}", started.elapsed()),
}
}
}
}输出（时间会有毫秒级抖动）：
fast => Ok(7)把 fast_source 也换成 silent_source，则 500ms 后两个分支各有一个超时结果，但 select! 只会取最先完成的那个，输出形如：
silent => 超时，耗时 500.3ms注意类型变化：分支的返回值从 Result<u8, E> 变成了 Result<Result<u8, E>, Elapsed>。内层是业务结果，外层是时间结果。这个双层 Result 是所有后续处理的核心，必须显式 match，不能 ? 一把梭，否则你会把超时和业务错误混为一谈。
共享总 deadline：剩下的时间才是每个分支的预算
上面每个分支各自写死 500ms 并不总是对的。更常见的需求是：上游给了本次调用 2 秒的总预算，四条数据源并行读，谁在 2 秒内没凑齐就整体失败。这时应该先算出 Deadline，再用 timeout_at（而不是 timeout）来约束分支，这样每个分支的绝对截止点一致，不会因为前一个分支已经消耗了时间而让后续分支多等。
use std::time::Duration;
use tokio::time::{timeout_at, Instant, sleep};
struct Reading {
source: &'static str,
value: Option<String>,
}
async fn source_slow() -> Result<String, &'static str> {
sleep(Duration::from_millis(1200)).await;
Ok("slow-payload".to_string())
}
async fn source_hang() -> Result<String, &'static str> {
sleep(Duration::from_secs(60)).await;
Ok("never".to_string())
}
async fn source_err() -> Result<String, &'static str> {
sleep(Duration::from_millis(60)).await;
Err("upstream 503")
}
#[tokio::main]
async fn main() {
// 上游给的总预算 800ms
let deadline = Instant::now() + Duration::from_millis(800);
let mut out: Vec<Reading> = Vec::new();
tokio::select! {
r = timeout_at(deadline, source_slow()) => {
out.push(Reading {
source: "slow",
value: r.ok().and_then(|x| x.ok()),
});
}
r = timeout_at(deadline, source_hang()) => {
out.push(Reading {
source: "hang",
value: r.ok().and_then(|x| x.ok()),
});
}
r = timeout_at(deadline, source_err()) => {
out.push(Reading {
source: "err",
value: r.ok().and_then(|x| x.ok()),
});
}
}
for r in &out {
println!("{} => {:?}", r.source, r.value);
}
}输出：
err => Nonesource_err 在 60ms 就返回了 Err("upstream 503")，所以 select! 立刻结束，slow 和 hang 两个分支的 future 被 drop。这是 select! 的既有行为，与是否套 timeout 无关。如果你需要「尽量收齐、只放弃超时的那一路」，那就必须把 select! 放进循环里，用可变的状态记录每一路是否已完成，见下一节。
这里有个易错点：timeout_at 的 deadline 必须是 tokio::time::Instant，不是 std::time::Instant。两者不通用，混用会得到编译错误：
error[E0308]: mismatched types
--> src/main.rs:31:26
|
31 | tokio::time::timeout_at(std_deadline, fut)
| ------------------------- ^^^^^^^^^^^^ expected `tokio::time::Instant`, found `std::time::Instant`
|
= note: expected struct `tokio::time::Instant`
found struct `std::time::Instant`修法有两种：换成 tokio::time::Instant::now() + dur，或者用 tokio::time::Instant::from_std(std_instant) 做一次显式转换。不要用 as 强转，那是编译不过的。另外，开启 tokio 的 test-util feature 后，tokio::time::Instant 会跟随被拨快的虚拟时钟，而 std::time::Instant 不会——这也是测试里 mock 时间必须用 tokio 类型的原因。
循环多路读取：用状态位区分「已完成」与「超时」
真实场景里更常见的是：三个传感器要读，各自有独立超时，最后一并返回，超时的那路标记为缺失而不是让整批失败。做法是把每路包装成一个 Pin<Box<dyn Future>> 并保存在 Option 里，select! 分支用 if 守卫在 future 为 Some 时才参与，完成后置 None 并从循环条件里排除。
use std::future::Future;
use std::pin::Pin;
use std::time::Duration;
use tokio::time::{timeout_at, Instant, sleep};
type BoxFut = Pin<Box<dyn Future<Output = Result<String, &'static str>> + Send>>;
#[derive(Debug)]
struct Slot {
name: &'static str,
fut: Option<BoxFut>,
value: Option<String>,
timed_out: bool,
}
fn sensor_a() -> BoxFut {
Box::pin(async {
sleep(Duration::from_millis(40)).await;
Ok("a=21.5".to_string())
})
}
fn sensor_b() -> BoxFut {
Box::pin(async {
sleep(Duration::from_millis(5000)).await;
Ok("b=99.9".to_string())
})
}
fn sensor_c() -> BoxFut {
Box::pin(async {
sleep(Duration::from_millis(30)).await;
Err("c: crc mismatch")
})
}
#[tokio::main]
async fn main() {
let deadline = Instant::now() + Duration::from_millis(300);
let mut slots = vec![
Slot { name: "a", fut: Some(sensor_a()), value: None, timed_out: false },
Slot { name: "b", fut: Some(sensor_b()), value: None, timed_out: false },
Slot { name: "c", fut: Some(sensor_c()), value: None, timed_out: false },
];
while slots.iter().any(|s| s.fut.is_some()) {
let (idx, outcome) = tokio::select! {
biased;
(i, r) = async {
// 0 号槽位
(0usize, timeout_at(deadline, slots[0].fut.as_mut().unwrap()).await)
}, if slots[0].fut.is_some() => (i, r),
(i, r) = async {
(1usize, timeout_at(deadline, slots[1].fut.as_mut().unwrap()).await)
}, if slots[1].fut.is_some() => (i, r),
(i, r) = async {
(2usize, timeout_at(deadline, slots[2].fut.as_mut().unwrap()).await)
}, if slots[2].fut.is_some() => (i, r),
};
let slot = &mut slots[idx];
slot.fut = None;
match outcome {
Ok(Ok(v)) => slot.value = Some(v),
Ok(Err(e)) => {
eprintln!("{} 读取失败: {}", slot.name, e);
}
Err(_) => slot.timed_out = true,
}
}
for s in &slots {
println!(
"{} => value={:?} timed_out={}",
s.name, s.value, s.timed_out
);
}
}预期输出：
c 读取失败: c: crc mismatch
a => value=Some("a=21.5") timed_out=false
b => value=None timed_out=true
c => value=None timed_out=false这个骨架里有三个必须讲清的点。
第一，as_mut().unwrap() 不会 panic。 因为每个分支都带 if slots[i].fut.is_some() 守卫，守卫在分支被选中之前求值；守卫为 false 的分支直接被禁用，不会进入表达式体。这是 select! 的语义保证。但守卫里的条件不能有副作用，它在每轮轮询中可能被多次求值（每次某个分支就绪时都会重新检查其他分支的守卫，取决于实现细节），写 slots.remove(0) 这种就会出问题。
第二，biased; 在这里是有意的。 因为它按书写顺序检查，a、c 这类快源会先被消费，b 这种慢源最后才被检查，避免了随机顺序下 b 先被选中、阻塞其他分支完成机会的低效。但要清楚 biased; 会让 0 号槽位长期占优；如果 a 是一个高频持续产出的源，b/c 会被饿死。本例中每个槽位完成一次就置 None，所以不存在饿死。
第三，如果用 timeout_at 共享同一个 deadline，那么「超时」是批量发生的。 上例中 b 在 300ms 被判定超时，但如果 a 和 c 都耗到 290ms，留给 b 的预算只有 10ms，这在业务上往往不合理。这时应该给每一路独立预算：timeout(slot.budget, fut)，再用外层一个 timeout_at 做总兜底。两层并存的写法下面给出。
两层超时：单路预算 + 整体兜底
单路预算负责「这一路太慢就丢掉」，整体兜底负责「整个函数不能超过上游给的 SLA」。两者叠加时，注意外层一定用 timeout_at 且 deadline 只计算一次，否则每轮循环都重新 now() + dur，兜底就变成了滚动窗口，永远不会触发。
use std::time::Duration;
use tokio::time::{sleep, timeout, timeout_at, Instant};
async fn read_sensor(name: &'static str, cost: Duration) -> Result<String, &'static str> {
sleep(cost).await;
Ok(format!("{name}=ok"))
}
async fn collect_all() -> Vec<(&'static str, Option<String>)> {
// 总 SLA：1 秒。只算一次
let hard_deadline = Instant::now() + Duration::from_millis(1000);
let mut results = Vec::new();
let tasks: [(&'static str, Duration); 3] = [
("t1", Duration::from_millis(100)),
("t2", Duration::from_millis(400)),
("t3", Duration::from_millis(900)),
];
for (name, cost) in tasks {
// 单路预算 500ms，再被总 deadline 截断
let per_call = timeout(Duration::from_millis(500), read_sensor(name, cost));
let r = timeout_at(hard_deadline, per_call).await;
let v = match r {
Ok(Ok(Ok(s))) => Some(s),
Ok(Ok(Err(e))) => {
eprintln!("{name} 业务错误: {e}");
None
}
Ok(Err(_)) => {
eprintln!("{name} 单路超时(500ms)");
None
}
Err(_) => {
eprintln!("{name} 被总 deadline 截断");
None
}
};
results.push((name, v));
}
results
}
#[tokio::main]
async fn main() {
let t0 = Instant::now();
let out = collect_all().await;
println!("{:?}", out);
println!("总耗时 {:?}", t0.elapsed());
}输出：
t1 => Some("t1=ok")
t2 => Some("t2=ok")
t3 单路超时(500ms)
t3 => None
总耗时 1.0002s注意这里是串行 for 循环，所以 t3 的 500ms 单路预算在 100+400=500ms 之后才开始计时，到 1000ms 时刚好被单路预算卡掉，同时硬 deadline 也是 1000ms，两者几乎同时到达。当单路预算与总 deadline 撞在一起时，timeout_at 外层先返回还是内层先返回取决于哪个 timer 先触发，属于实现细节，业务逻辑上应把二者当作「都可能发生」来处理——所以上面三层 match 里的 Ok(Err(_)) 和 Err(_) 必须都写，不能只写一个。
顺带说一下内联等待的写法。很多人喜欢用 tokio::join! 让三路真正并行，这时每一路的 timeout 依然独立生效：
use tokio::time::{sleep, timeout, Duration};
async fn fn_a() -> &'static str {
sleep(Duration::from_millis(120)).await;
"a"
}
async fn fn_b() -> &'static str {
sleep(Duration::from_secs(30)).await;
"b"
}
#[tokio::main]
async fn main() {
let (ra, rb) = tokio::join!(
timeout(Duration::from_millis(300), fn_a()),
timeout(Duration::from_millis(300), fn_b()),
);
println!("a={:?} b={:?}", ra.ok(), rb.ok());
}输出 a=Some("a") b=None，总耗时约 300ms。join! 会等所有 future 完成，所以总耗时是两路中较慢的那个；select! 抢占则只要一路完成就返回。选哪个取决于业务：是要最快的那一路，还是要全部结果。
必须注意的三个坑
取消安全性（cancellation safety）
把 timeout 套到分支上，本质是在超时的瞬间 drop 掉内部 future。所以被包裹的 future 必须是取消安全的，否则丢数据。AsyncReadExt::read、AsyncWriteExt::write_all、tokio::sync::mpsc::Receiver::recv、tokio::sync::oneshot::Receiver 是取消安全的（drop 后再调用不会丢已读数据）；AsyncBufReadExt::lines 的 next_line、BufReader 上的 read_line、以及自己手写的「先发请求再等响应再写状态」的复合 future 不是。对后者，超时 drop 后你已经发出的请求仍然在对端执行，重试会造成重复提交。
tokio 文档里有一条必须记住的结论：select! 在分支被 drop 时会丢弃该分支已经产生但尚未返回的部分结果。因此对 TcpStream 做带超时的读取时，更稳的写法是用 tokio::io::AsyncReadExt::read 读固定长度的 buffer 反复累积，而不是用 lines() 等整行。这类 bug 的现象是：偶发地少一行、少一个包，日志里却看不到任何错误。
timeout 不会中断阻塞任务
如果分支里跑的是 tokio::task::spawn_blocking 包住的同步代码，timeout 只是让等待方返回，blocking 线程池里的任务仍会跑完，且不能被取消。现象是超时日志打了，但 CPU 还在涨。std::thread::sleep 放进 async 分支同理——它会让整个 worker 线程卡住，超时回调根本没有机会执行。所有耗时操作必须用 tokio 的异步版本，或至少放在 spawn_blocking 里并接受「不可取消」。
Mock 时间要做对
用 tokio::time::pause() / advance() 做单测时，#[tokio::test] 需要开启 test-util feature，且只能用 tokio::time::sleep。若混入 std::thread::sleep 或 std::time::Instant，虚拟时钟不会推进，测试要么挂死，要么出现「1 毫秒的 timeout 测了 5 秒」的怪象。Cargo.toml 写法：
[dependencies]
tokio = { version = "1.40", features = ["macros", "rt-multi-thread", "time", "sync", "io-util", "net", "test-util"] }另外 tokio::time::timeout 内部使用的 timer 精度约为 1ms，且受 runtime 当前负载影响，不要写断言 assert_eq!(elapsed, Duration::from_millis(500))，应写成 assert!(elapsed >= Duration::from_millis(500))，并给上界留足余量（例如 < 650ms），否则 CI 上偶发失败。
小结
把 timeout 与 select! 结合的关键不在语法，而在三件事：一是明确 select! 自身没有时间语义，超时必须由分支内的 future 承担；二是搞清 timeout 与 timeout_at 的区别，需要共享预算时先算一次 Instant 再 timeout_at，绝不能在循环里反复 now() + dur；三是接受超时即 drop 的事实，只对取消安全的 future 这么干，对非取消安全的操作用状态标注或 join! + 独立超时来替代抢占。返回值的 Result<Result<T, E>, Elapsed> 双层结构应当在类型上就区分开，而不是靠日志猜。
FAQ
为什么 timeout 包裹后返回类型变成了两层 Result，有没有更清爽的写法？
外层 Result 的 Err 类型固定是 tokio::time::error::Elapsed（零大小类型），内层是分支自己的错误。想清爽可以封装一个枚举：
use std::fmt;
use tokio::time::Elapsed;
#[derive(Debug)]
enum ReadError<E> {
Timeout,
Inner(E),
}
impl<E: fmt::Display> fmt::Display for ReadError<E> {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
ReadError::Timeout => write!(f, "read timed out"),
ReadError::Inner(e) => write!(f, "inner error: {e}"),
}
}
}
fn flatten<T, E>(r: Result<Result<T, E>, Elapsed>) -> Result<T, ReadError<E>> {
match r {
Ok(Ok(v)) => Ok(v),
Ok(Err(e)) => Err(ReadError::Inner(e)),
Err(_) => Err(ReadError::Timeout),
}
}之后在分支里写 flatten(r) 即可，且这个枚举可以直接 impl std::error::Error 或交给 thiserror 生成。
select! 里同时用了 timeout 和 else，为什么 else 从没执行过？
因为只要还有一个分支的 if 守卫为 true，else 就不会执行。timeout 分支本身会在超时后正常完成并返回 Err(Elapsed)，这被算作「分支完成」，不是「所有分支禁用」。else 只服务于全部守卫为 false 的场景，与时间无关。
报错 error[E0599]: no method named `as_mut` found for enum `Option<...> 怎么解？
因为 slots[i].fut 是 Option 而你写成了 slots[i].fut.as_mut() 之外的形式（例如对 &mut Option 调用了 Pin::new）。正确做法是先 let fut = slots[i].fut.as_mut().unwrap(); 拿到 &mut BoxFut，再解引用成 &mut (dyn Future)：timeout_at(deadline, &mut **fut)。另一种常见原因是没有引入 use std::pin::Pin;。
超时后连接没有被真正关闭，对端还能收到后续请求？
timeout 只 drop future，不 drop TcpStream——如果你把 &mut stream 传进了分支，stream 的所有权还在外层，连接不受影响。这是设计上的优点（可以复用连接），但也意味着超时的语义是「我不等了」，而不是「我取消了」。若业务要求真取消，必须在超时分支里显式调用 stream.shutdown().await，或改用 spawn 出任务后 handle.abort()，后者仅对 abort 能打断的 await 点有效，同样不保证对端感知。
为什么我的 timeout(Duration::from_millis(100), fut) 实际等了三四秒才返回？
两个典型原因。其一是 fut 在执行过程中调用了阻塞函数（std::thread::sleep、同步 IO、block_on），把 worker 线程占死了，timer 驱动无法及时执行。其二是使用了 #[tokio::main] 的 current_thread 变体而某处又 spawn_blocking，导致调度队列积压。解决方向是一致的：让 await 链上不出现阻塞调用，必要时用 tokio-console 观察任务的 poll 耗时。开启方式是在 Cargo.toml 加 console-subscriber = "0.4"、运行时 feature 加 tokio_unstable，然后用 RUSTFLAGS="--cfg tokio_unstable" cargo run。