Rust 中的并发与异步：高效执行多个 Future 的技巧与方法

在 Rust 中，异步编程已成为高效处理并发任务的重要工具。理解并发操作如何在多个 Future 之间有效协调，对于编写高性能、响应迅速的程序至关重要。本篇文章将详细介绍如何在 Rust 中同时执行多个 Future，并深入探讨常见的异步工具和技术，如 join!、select! 和 FutureUnordered，帮助开发者更好地掌握异步编程的精髓。

本文介绍了 Rust 中几种执行多个异步操作的方法，包括 join!、select!、try_join! 和 FutureUnordered。通过这些技术，开发者可以高效地处理并发任务，确保异步操作的正确性和性能。文章详细讲解了如何使用这些工具来执行多个异步操作，同时保证代码的简洁和可读性，涵盖了错误处理、select! 的高级用法及其与 Unpin 和 FusedFuture trait 的交互。

Executing Multiple Futures at a Time（同时执行多个 Future）

本节内容

• 真正的异步应用程序通常需要同时执行几个不同的操作
• 介绍一些可同时执行多个异步操作的方式：
• Join!，等待所有 Future 完成
• Select!，等待多个 future 中的一个完成
• Spawning，创建一个顶级任务，他会运行一个 future 直至完成
• FutureUnordered，一组 Future，它们会产生每个子 Future 的结果

1. join

join

• futures::join 宏，它使得在等待多个 future完成的时候，可以同时并发的执行它们。

例子

async fn get_book_and_music() -> (Book, Music) {
    let book = get_book().await;
    let music = get_music().await;
    (book, music)
}

要支持同时看书和听歌，有些人可能会生成下面代码：

// WRONG -- don't do this
async fn get_book_and_music() -> (Book, Music) {
    let book_future = get_book();
    let music_future = get_music();
    (book_future.await, music_future.await)
}

为了正确的并发运行两个 Future ， 我们来试试 futures::join! 宏:

use futures::join;

async fn get_book_and_music() -> (Book, Music) {
    let book_fut = get_book();
    let music_fut = get_music();
    join!(book_fut, music_fut)
}

try_join

• 对于返回 Result 的 future，更考虑使用 try_join!
• 如果子 future 中某一个返回了错误，try_join! 会立即完成

例子

use futures::try_join;

async fn get_book() -> Result<Book, String> { /* ... */ Ok(Book) }
async fn get_music() -> Result<Music, String> { /* ... */ Ok(Music) }

async fn get_book_and_music() -> Result<(Book, Music), String> {
    let book_fut = get_book();
    let music_fut = get_music();
    try_join!(book_fut, music_fut)
}

有一点需要注意，传给 try_join! 的所有 Future 都必须拥有相同的错误类型。如果错误类型不同，可以考虑使用来自 futures::future::TryFutureExt 模块的 map_err和err_info方法将错误进行转换:

use futures::{
    future::TryFutureExt,
    try_join,
};

async fn get_book() -> Result<Book, ()> { /* ... */ Ok(Book) }
async fn get_music() -> Result<Music, String> { /* ... */ Ok(Music) }

async fn get_book_and_music() -> Result<(Book, Music), String> {
    let book_fut = get_book().map_err(|()| "Unable to get book".to_string());
    let music_fut = get_music();
    try_join!(book_fut, music_fut)
}

2. select

select

• futures::select 宏可同时运行多个 future，允许用户在任意一个 future 完成时进行响应

use futures::{
    future::FutureExt, // for `.fuse()`
    pin_mut,
    select,
};

async fn task_one() { /* ... */ }
async fn task_two() { /* ... */ }

async fn race_tasks() {
    let t1 = task_one().fuse();
    let t2 = task_two().fuse();

    pin_mut!(t1, t2);

    select! {
        () = t1 => println!("task one completed first"),
        () = t2 => println!("task two completed first"),
    }
}

default => ... 和 complete =>

• select 支持 default 和 complete 分支
• default：如果选中的 future 尚未完成，就会运行 default 分支
• 拥有 default 的 select 总是会立即返回
• complete：它用于所有选中的 future 都已完成的情况。
• complete 分支当所有的 Future 和 Stream 完成后才会被执行，它往往配合loop使用，loop用于循环完成所有的 Future
• default分支，若没有任何 Future 或 Stream 处于 Ready 状态， 则该分支会被立即执行

例子

use futures::{future, select};

async fn count() {
    let mut a_fut = future::ready(4);
    let mut b_fut = future::ready(6);
    let mut total = 0;

    loop {
        select! {
            a = a_fut => total += a,
            b = b_fut => total += b,
            complete => break,
            default => unreachable!(), // never runs (futures are ready, then complete)   该分支永远不会运行，因为`Future`会先运行，然后是`complete`
        };
    }
    assert_eq!(total, 10);
}

与 Unpin 和 FusedFuture 交互

• 前面的例子中，需要在返回的 future 上调用 .fuse()，也调用了 pin_mut。
• 因为 select 里面的 future 必须使用 Unpin 和 FusedFuture 这两个 trait。
• 必须 Unpin：select 使用的 future 不是按值的，而是按可变引用。
• 未完成的 future 在调用 select 后 仍可使用
• 必须 FusedFuture：在 future 完成后，select 不可以对它进行 poll
• 实现 FusedFuture 的 future 会追踪其完成状态，这样在 select 循环里，就只会 poll 没有完成的 future

Stream 也有 FusedStream trait

首先，.fuse()方法可以让 Future 实现 FusedFuture 特征， 而 pin_mut! 宏会为 Future 实现 Unpin特征，这两个特征恰恰是使用 select 所必须的:

• Unpin，由于 select 不会通过拿走所有权的方式使用Future，而是通过可变引用的方式去使用，这样当 select 结束后，该 Future 若没有被完成，它的所有权还可以继续被其它代码使用。
• FusedFuture的原因跟上面类似，当 Future 一旦完成后，那 select 就不能再对其进行轮询使用。Fuse意味着熔断，相当于 Future 一旦完成，再次调用poll会直接返回Poll::Pending。

只有实现了FusedFuture，select 才能配合 loop 一起使用。假如没有实现，就算一个 Future 已经完成了，它依然会被 select 不停的轮询执行。

Stream 稍有不同，它们使用的特征是 FusedStream。 通过.fuse()(也可以手动实现)实现了该特征的 Stream，对其调用.next() 或 .try_next()方法可以获取实现了FusedFuture特征的Future:

use futures::{
    stream::{Stream, StreamExt, FusedStream},
    select,
};

async fn add_two_streams(
    mut s1: impl Stream<Item = u8> + FusedStream + Unpin,
    mut s2: impl Stream<Item = u8> + FusedStream + Unpin,
) -> u8 {
    let mut total = 0;

    loop {
        let item = select! {
            x = s1.next() => x,
            x = s2.next() => x,
            complete => break,
        };
        if let Some(next_num) = item {
            total += next_num;
        }
    }

    total
}

select 循环里使用 Fuse 和 FuturesUnordered

• Fuse::terminated()，允许构建空的、已完成的 future，后续可以为它填充一个需要运行的 future
• 适用于在 select 循环里产生且需要在这运行的任务，这种场景
• 当同个 future 的多个副本需同时运行时，使用 FuturesUnordered 类型

use futures::{
    future::{Fuse, FusedFuture, FutureExt},
    stream::{FusedStream, Stream, StreamExt},
    pin_mut,
    select,
};

async fn get_new_num() -> u8 { /* ... */ 5 }

async fn run_on_new_num(_: u8) { /* ... */ }

async fn run_loop(
    mut interval_timer: impl Stream<Item = ()> + FusedStream + Unpin,
    starting_num: u8,
) {
    let run_on_new_num_fut = run_on_new_num(starting_num).fuse();
    let get_new_num_fut = Fuse::terminated();
    pin_mut!(run_on_new_num_fut, get_new_num_fut);
    loop {
        select! {
            () = interval_timer.select_next_some() => {
                // The timer has elapsed. Start a new `get_new_num_fut`
                // if one was not already running.
                // 定时器已结束，若`get_new_num_fut`没有在运行，就创建一个新的
                if get_new_num_fut.is_terminated() {
                    get_new_num_fut.set(get_new_num().fuse());
                }
            },
            new_num = get_new_num_fut => {
                // A new number has arrived -- start a new `run_on_new_num_fut`,
                // dropping the old one.
                // 收到新的数字 -- 创建一个新的`run_on_new_num_fut`并丢弃掉旧的
                run_on_new_num_fut.set(run_on_new_num(new_num).fuse());
            },
            // Run the `run_on_new_num_fut` // 运行 `run_on_new_num_fut`
            () = run_on_new_num_fut => {},
            // panic if everything completed, since the `interval_timer` should
            // keep yielding values indefinitely.
            // 若所有任务都完成，直接 `panic`， 原因是 `interval_timer` 应该连续不断的产生值，而不是结束          //后，执行到 `complete` 分支
            complete => panic!("`interval_timer` completed unexpectedly"),
        }
    }
}

当某个 Future 有多个拷贝都需要同时运行时，可以使用 FuturesUnordered 类型。下面的例子跟上个例子大体相似，但是它会将 run_on_new_num_fut 的每一个拷贝都运行到完成，而不是像之前那样一旦创建新的就终止旧的。

use futures::{
    future::{Fuse, FusedFuture, FutureExt},
    stream::{FusedStream, FuturesUnordered, Stream, StreamExt},
    pin_mut,
    select,
};

async fn get_new_num() -> u8 { /* ... */ 5 }

async fn run_on_new_num(_: u8) -> u8 { /* ... */ 5 }

// 使用从 `get_new_num` 获取的最新数字 来运行 `run_on_new_num`
//
// 每当计时器结束后，`get_new_num` 就会运行一次，它会立即取消当前正在运行的`run_on_new_num` ,
// 并且使用新返回的值来替换 
async fn run_loop(
    mut interval_timer: impl Stream<Item = ()> + FusedStream + Unpin,
    starting_num: u8,
) {
    let mut run_on_new_num_futs = FuturesUnordered::new();
    run_on_new_num_futs.push(run_on_new_num(starting_num));
    let get_new_num_fut = Fuse::terminated();
    pin_mut!(get_new_num_fut);
    loop {
        select! {
            () = interval_timer.select_next_some() => {
                // The timer has elapsed. Start a new `get_new_num_fut`
                // if one was not already running.
                // 定时器已结束，若`get_new_num_fut`没有在运行，就创建一个新的
                if get_new_num_fut.is_terminated() {
                    get_new_num_fut.set(get_new_num().fuse());
                }
            },
            new_num = get_new_num_fut => {
                // A new number has arrived -- start a new `run_on_new_num_fut`.
                // 收到新的数字 -- 创建一个新的`run_on_new_num_fut` (并没有像之前的例子那样丢弃掉旧值)
                run_on_new_num_futs.push(run_on_new_num(new_num));
            },
            // Run the `run_on_new_num_futs` and check if any have completed
            // 运行 `run_on_new_num_futs`, 并检查是否有已经完成的
            res = run_on_new_num_futs.select_next_some() => {
                println!("run_on_new_num_fut returned {:?}", res);
            },
            // panic if everything completed, since the `interval_timer` should
            // keep yielding values indefinitely.
           // 若所有任务都完成，直接 `panic`， 原因是 `interval_timer` 应该连续不断的产生值，而不是结束
            //后，执行到 `complete` 分支
            complete => panic!("`interval_timer` completed unexpectedly"),
        }
    }
}

总结

异步编程使得 Rust 能够处理高并发场景，保证程序在执行多个任务时仍然保持高效和响应性。通过 join!、select! 等工具，开发者可以灵活地管理多个 Future 的执行顺序与并发执行策略。了解这些技术，并在适当的场合应用它们，将使得 Rust 开发者能够编写出更具性能和可靠性的异步应用。