async Rust 的取消安全

845 字

4 分钟

async Rust 的取消安全

2026-07-27

Rust

/

异步

/

并发

Rust 的 future 是惰性的：创建它不会做任何事，只有 poll 才推进。取消也不是一个显式的 cancel() 调用，而是把 future drop 掉。tokio::select! 在第一个分支完成后丢弃其余 future，这就是最常见的取消来源。

取消点在哪里#

每个 .await 都是一个潜在的中断点。编译器把 async fn 展开成状态机，挂起时把局部变量保存在状态里；被 drop 时，已经执行到一半的操作不会自动回滚。

1
async fn copy(mut src: impl AsyncRead + Unpin, mut dst: impl AsyncWrite + Unpin) -> io::Result<()> {
2
    let mut buf = [0u8; 8192];
3
    loop {
4
        let n = src.read(&mut buf).await?; // 取消点
5
        if n == 0 { return Ok(()); }
6
        dst.write_all(&buf[..n]).await?;   // 取消点
7
    }
8
}

如果在 write_all 中途被取消，已经写出去的部分字节不会撤回。调用方如果重试，就会重复发送。这段代码在“取消后丢弃整个连接”的语义下没问题，但在 select! 里复用同一个流就有问题。

哪些操作是取消安全的#

一个经验规则：在单个 await 点完成的操作通常是取消安全的；需要多次 await 才能完成一个逻辑单元的操作，通常不是。

tokio::sync::mpsc::Receiver::recv 是取消安全的：被取消时没有消息被取走。oneshot::Receiver 也是：要么收到，要么没收到。tokio::sync::Mutex::lock 在取消时会释放排队位置，也不会留下持锁状态。

而 AsyncReadExt::read_exact 不是：它内部循环多次 read，取消时缓冲区里可能已经有部分数据，下次调用会从新的位置开始，导致数据错位。write_all 同理。

1
// 不取消安全：中途取消会丢失已读的字节
2
let mut header = [0u8; 16];
3
reader.read_exact(&mut header).await?;

要让它可取消，要么把整个操作放进一个独立 task，取消时连带丢弃连接；要么自己维护偏移量，把“读了多少”变成显式状态。

select! 的两种写法#

1
// 写法一：分支里只做一件事，取消是安全的
2
tokio::select! {
3
    msg = rx.recv() => handle(msg),
4
    _ = shutdown.cancelled() => return,
5
}
6

7
// 写法二：分支里是一个需要多次 await 的复合操作
8
tokio::select! {
9
    _ = copy(src, dst) => {}
10
    _ = shutdown.cancelled() => {
11
        // 这里 copy 已经被 drop，dst 处于半写状态
12
    }
13
}

第二种写法不是不能用，但取消之后不能假设 dst 仍然处于一致状态。常见做法是取消时直接关闭连接，而不是继续复用。

用 task 隔离取消#

如果一段逻辑必须在后台完成、不受调用方取消影响，就 tokio::spawn 出去，用 JoinHandle 等待结果。spawn 出来的 task 只有运行时关闭才会被取消。

1
let handle = tokio::spawn(async move {
2
    copy(src, dst).await
3
});
4

5
tokio::select! {
6
    res = handle => res??,
7
    _ = shutdown.cancelled() => {
8
        // 可以选择 abort，也可以让它自己跑完
9
        // handle.abort();
10
    }
11
}