宏 select 

macro_rules! select {
    {
        $(
            biased;
        )?
        $(
            $bind:pat = $fut:expr $(, if $cond:expr)? => $handler:expr,
        )*
        $(
            else => $els:expr $(,)?
        )?
    } => { ... };
}

等待多个并发分支，当 第一个 分支完成时返回，取消剩余的分支。

select! 宏必须在 async 函数、闭包和代码块内部使用。

select! 宏接受一个或多个具有以下模式的分支：

<pattern> = <async expression> (, if <precondition>)? => <handler>,

此外，select! 宏可以包含一个可选的 else 分支，当没有其他分支匹配其模式时该分支会被求值：

else => <expression>

该宏会汇总所有 <async expression> 表达式，并在当前 任务上并发地运行它们。一旦首个表达式 以匹配其 <pattern> 的值完成，select! 宏 就会返回对已完成分支的 <handler> 表达式求值的结果。

此外，每个分支可以包含一个可选的 if 前置条件。如果 前置条件返回 false，则该分支会被禁用。提供的 <async expression> 仍会被求值，但生成的 future 永远不会被 poll。在循环中使用 select! 时，这一能力非常有用。

select! 表达式的完整生命周期如下：

1. Evaluate all provided <precondition> expressions. If the precondition returns false, disable the branch for the remainder of the current call to select!. Re-entering select! due to a loop clears the “disabled” state.
2. Aggregate the <async expression>s from each branch, including the disabled ones. If the branch is disabled, <async expression> is still evaluated, but the resulting future is not polled.
3. If all branches are disabled: go to step 6.
4. Concurrently await on the results for all remaining <async expression>s.
5. Once an <async expression> returns a value, attempt to apply the value to the provided <pattern>. If the pattern matches, evaluate the <handler> and return. If the pattern does not match, disable the current branch for the remainder of the current call to select!. Continue from step 3.
6. Evaluate the else expression. If no else expression is provided, panic.

Runtime characteristics

通过在当前任务上运行所有 async 表达式，这些表达式 能够并发运行，但不能并行运行。这意味着所有 表达式都在同一线程上运行，如果某个分支阻塞了该线程， 所有其他表达式将无法继续。如果需要并行执行， 可以使用 tokio::spawn 生成每个 async 表达式，并将 join handle 传给 select!。

Fairness

默认情况下，select! 会随机选择一个分支优先检查。这 在循环中调用 select! 且分支总是 ready 的情况下提供了一定程度的公平性。

可以通过在宏使用的开头添加 biased; 来覆盖此行为。 详情请参阅示例。这会使 select 按照 future 从上到下出现的顺序对其进行 poll。出于以下几个原因，你可能 希望这样做：

• The random number generation of tokio::select! has a non-zero CPU cost
• Your futures may interact in a way where known polling order is significant

但在这种模式下有一个重要的注意事项。确保 future 的 poll 顺序是公平的就成了你的 责任。例如，如果你要在流和关闭 future 之间进行 select，且该流 有大量消息且消息之间几乎没有时间间隔， 那么应该把关闭 future 放在 select! 列表的更前面， 以确保它始终被 poll，而不会因流一直 ready 而被忽略。

Panics

如果所有分支都被禁用且没有提供 else 分支，select! 宏会 panic。分支在所提供的 if 前置条件返回 false 时被禁用， 或者当模式不匹配 <async expression> 的结果时也会被禁用。

Cancellation safety

在循环中使用 select! 从多个来源接收消息时， 应确保接收调用是 cancellation safe 的，以避免 丢失消息。本节介绍各种常见方法， 并说明它们是否可安全取消。本节中的列表并不 详尽。

以下方法是可取消安全的：

• tokio::sync::mpsc::Receiver::recv
• tokio::sync::mpsc::UnboundedReceiver::recv
• tokio::sync::broadcast::Receiver::recv
• tokio::sync::watch::Receiver::changed
• tokio::net::TcpListener::accept
• tokio::net::UnixListener::accept
• tokio::signal::unix::Signal::recv
• tokio::io::AsyncReadExt::read on any AsyncRead
• tokio::io::AsyncReadExt::read_buf on any AsyncRead
• tokio::io::AsyncWriteExt::write on any AsyncWrite
• tokio::io::AsyncWriteExt::write_buf on any AsyncWrite
• tokio_stream::StreamExt::next on any Stream
• futures::stream::StreamExt::next on any Stream

以下方法不是可取消安全的，可能导致数据丢失：

• tokio::io::AsyncReadExt::read_exact
• tokio::io::AsyncReadExt::read_to_end
• tokio::io::AsyncReadExt::read_to_string
• tokio::io::AsyncWriteExt::write_all

以下方法不可安全取消，因为它们使用队列来实现 公平性，取消会丢失在队列中的位置：

• tokio::sync::Mutex::lock
• tokio::sync::RwLock::read
• tokio::sync::RwLock::write
• tokio::sync::Semaphore::acquire
• tokio::sync::Notify::notified

要判断你自己的方法是否可安全取消，请查找 .await 的使用位置。这是因为当异步方法 被取消时，总是发生在某个 .await 处。如果你的函数 即便在 .await 处等待时重启也能正确 执行，那么它就是可安全取消的。

可以用以下方式定义取消安全性： 如果你有一个 尚未完成的 future， 那么丢弃该 future 并重新创建它 必须是一个空操作。 这个定义源于 在循环中使用 select! 的场景。 如果没有这个保证， 当另一个分支完成时你将通过循环重新启动 select!， 此时你将丢失进度。

请注意，取消一个不可安全取消的操作并不 一定是错误的。例如，如果因为 应用程序正在关闭而取消某个任务，那么 可能并不在意部分读取的数据丢失。

示例

两分支的基本 select。

async fn do_stuff_async() {
    // async work
}

async fn more_async_work() {
    // more here
}

tokio::select! {
    _ = do_stuff_async() => {
        println!("do_stuff_async() completed first")
    }
    _ = more_async_work() => {
        println!("more_async_work() completed first")
    }
};

基本的流选择。

use tokio_stream::{self as stream, StreamExt};

let mut stream1 = stream::iter(vec![1, 2, 3]);
let mut stream2 = stream::iter(vec![4, 5, 6]);

let next = tokio::select! {
    v = stream1.next() => v.unwrap(),
    v = stream2.next() => v.unwrap(),
};

assert!(next == 1 || next == 4);

收集两个流的内容。在本例中，我们依赖于模式 匹配以及 stream::iter 是“fused”的事实，即流 完成后，对 next() 的所有调用都会返回 None。

use tokio_stream::{self as stream, StreamExt};

let mut stream1 = stream::iter(vec![1, 2, 3]);
let mut stream2 = stream::iter(vec![4, 5, 6]);

let mut values = vec![];

loop {
    tokio::select! {
        Some(v) = stream1.next() => values.push(v),
        Some(v) = stream2.next() => values.push(v),
        else => break,
    }
}

values.sort();
assert_eq!(&[1, 2, 3, 4, 5, 6], &values[..]);

在多个 select! 表达式中使用同一个 future，可以通过 传递该 future 的引用来实现。这要求 future 是 Unpin 的。 通过 Box::pin 或栈固定，可以使 future 成为 Unpin。

此处一个 stream 最多被消耗 1 秒。

use tokio_stream::{self as stream, StreamExt};
use tokio::time::{self, Duration};

let mut stream = stream::iter(vec![1, 2, 3]);
let sleep = time::sleep(Duration::from_secs(1));
tokio::pin!(sleep);

loop {
    tokio::select! {
        maybe_v = stream.next() => {
            if let Some(v) = maybe_v {
                println!("got = {}", v);
            } else {
                break;
            }
        }
        _ = &mut sleep => {
            println!("timeout");
            break;
        }
    }
}

使用 select! 合并两个值。

use tokio::sync::oneshot;

let (tx1, mut rx1) = oneshot::channel();
let (tx2, mut rx2) = oneshot::channel();

tokio::spawn(async move {
    tx1.send("first").unwrap();
});

tokio::spawn(async move {
    tx2.send("second").unwrap();
});

let mut a = None;
let mut b = None;

while a.is_none() || b.is_none() {
    tokio::select! {
        v1 = (&mut rx1), if a.is_none() => a = Some(v1.unwrap()),
        v2 = (&mut rx2), if b.is_none() => b = Some(v2.unwrap()),
    }
}

let res = (a.unwrap(), b.unwrap());

assert_eq!(res.0, "first");
assert_eq!(res.1, "second");

使用 biased; 模式来控制 poll 顺序。

let mut count = 0u8;

loop {
    tokio::select! {
        // If you run this example without `biased;`, the polling order is
        // pseudo-random, and the assertions on the value of count will
        // (probably) fail.
        biased;

        _ = async {}, if count < 1 => {
            count += 1;
            assert_eq!(count, 1);
        }
        _ = async {}, if count < 2 => {
            count += 1;
            assert_eq!(count, 2);
        }
        _ = async {}, if count < 3 => {
            count += 1;
            assert_eq!(count, 3);
        }
        _ = async {}, if count < 4 => {
            count += 1;
            assert_eq!(count, 4);
        }

        else => {
            break;
        }
    };
}

避免存在竞争条件的 if 前置条件

由于使用 if 前置条件来禁用 select! 分支， 因此必须谨慎以避免遗漏值。

例如，下面是对 sleep 与 if 的不正确用法。目标是 重复运行一个异步任务，持续时间不超过 50 毫秒。 但有可能错过 sleep 完成的事件。

use tokio::time::{self, Duration};

async fn some_async_work() {
    // do work
}

let sleep = time::sleep(Duration::from_millis(50));
tokio::pin!(sleep);

while !sleep.is_elapsed() {
    tokio::select! {
        _ = &mut sleep, if !sleep.is_elapsed() => {
            println!("operation timed out");
        }
        _ = some_async_work() => {
            println!("operation completed");
        }
    }
}

panic!("This example shows how not to do it!");

在上面的示例中，即便 sleep.poll() 始终没有返回 Ready，sleep.is_elapsed() 仍可能返回 true。 这会引发潜在的竞态条件：当 sleep 在 while !sleep.is_elapsed() 检查和 select! 调用之间到期时，some_async_work() 会被不间断地运行，即便 sleep 已经过去。

一种在不发生竞争的情况下写此例的方式是：

use tokio::time::{self, Duration};

async fn some_async_work() {
    // do work
}

let sleep = time::sleep(Duration::from_millis(50));
tokio::pin!(sleep);

loop {
    tokio::select! {
        _ = &mut sleep => {
            println!("operation timed out");
            break;
        }
        _ = some_async_work() => {
            println!("operation completed");
        }
    }
}

Alternatives from the Ecosystem

select! 宏是管理多个异步 分支的强大工具，使任务能够在同一线程内并发运行。然而， 它的使用会引入一些挑战，尤其是在取消安全方面， 这可能导致难以察觉且难以调试的错误。在许多用例中， 生态中的替代方案可能更可取，因为它们通过提供 更清晰的语法、更可预测的控制流，以及减少对手动 处理 fuse 语义或取消安全等问题的需要，缓解了这些问题。

合并流

对于使用 loop { select! { ... } } 来 poll 多个任务的 情况，流合并提供了一种简洁的替代方案，天然支持安全的 取消处理，消除了数据丢失的风险。tokio_stream、 futures::stream 和 futures_concurrency 等库提供了合并 流并按顺序处理其输出的工具。

使用 select! 的示例

struct File;
struct Channel;
struct Socket;

impl Socket {
    async fn read_packet(&mut self) -> Vec<u8> {
        vec![]
    }
}

async fn read_send(_file: &mut File, _channel: &mut Channel) {
    // do work that is not cancel safe
}

// open our IO types
let mut file = File;
let mut channel = Channel;
let mut socket = Socket;

loop {
    tokio::select! {
        _ = read_send(&mut file, &mut channel) => { /* ... */ },
        _data = socket.read_packet() => { /* ... */ }
        _ = futures::future::ready(()) => break
    }
}

改用 merge

通过使用 merge，可以将多个异步任务统一为一个流， 无需手动管理任务，并降低数据丢失等 意外行为的风险。

use std::pin::pin;

use futures::stream::unfold;
use tokio_stream::StreamExt;

struct File;
struct Channel;
struct Socket;

impl Socket {
    async fn read_packet(&mut self) -> Vec<u8> {
        vec![]
    }
}

async fn read_send(_file: &mut File, _channel: &mut Channel) {
    // do work that is not cancel safe
}

enum Message {
    Stop,
    Sent,
    Data(Vec<u8>),
}

// open our IO types
let file = File;
let channel = Channel;
let socket = Socket;

let a = unfold((file, channel), |(mut file, mut channel)| async {
    read_send(&mut file, &mut channel).await;
    Some((Message::Sent, (file, channel)))
});
let b = unfold(socket, |mut socket| async {
    let data = socket.read_packet().await;
    Some((Message::Data(data), socket))
});
let c = tokio_stream::iter([Message::Stop]);

let mut s = pin!(a.merge(b).merge(c));
while let Some(msg) = s.next().await {
    match msg {
        Message::Data(_data) => { /* ... */ }
        Message::Sent => continue,
        Message::Stop => break,
    }
}

Future 竞速

如果需要在多个异步任务中等待 最先完成的一个， futures、 futures-lite 或 futures-concurrency 等生态工具提供了 用于 Future 竞速的简洁语法：

• futures_concurrency::future::Race
• futures::select
• futures::stream::select_all (for streams)
• futures_lite::future::or
• futures_lite::future::race

use futures_concurrency::future::Race;

let task_a = async { Ok("ok") };
let task_b = async { Err("error") };
let result = (task_a, task_b).race().await;

match result {
    Ok(output) => println!("First task completed with: {output}"),
    Err(err) => eprintln!("Error occurred: {err}"),
}