select!

futures::select宏同时跑多个future,允许用户在任意future完成时响应:


# #![allow(unused_variables)]
#fn main() {
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"),
    }
}
#}

上面的函数会并发跑t1t2。当t1t2结束时,对应的句柄(handler)会调用println!, 然后函数就会结束而不会完成剩下的任务。

select的基本格式为<pattern> = <expression> => <code>,,可以重复你想select的 任意多future。

default => ... and complete => ...

select也支持defaultcomplete分支。

default会在没有被select的future完成时执行,因此,带有default分支的select总是 马上返回,因为default会在没有其它future准备好的时候返回。

complete分支则用来处理所有被select的future都完成并且不需进一步处理的情况。这在循环 select时很好用:


# #![allow(unused_variables)]
#fn main() {
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)
        };
    }
    assert_eq!(total, 10);
}
#}

UnpinFusedFuture交互

你会注意到,在上面第一个例子中,我们在两个async fn函数返回的future上调用了.fuse(), 然后用pin_mut来固定他们。这两个调用都是必需的,用在select中的future必须实现UnpinFusedFuture

需要Unpin是因为select是用可变引用访问future的,不获取future的所有权。未完成的future 因此可以在select调用后继续使用。

类似的,需要FusedFuture是因为select一定不能轮询已完成的future。FusedFuture用来 实现追踪(track)future是否已完成。这种使得在循环中使用select成为客官您,只轮询尚未 完成的future。这可以从上面的例子中看出,a_futb_fut可能会在第二次循环的时候已经 完成了。因为future::ready返回的future实现了FusedFuture,所以select可以知道不必 再次轮询它了。

注意,stream也有对应的FusedStream特质。实现了这个特质或者被.fuse()包装的Stream会 从它们的.next/try_next()组合子中返还FusedFutre


# #![allow(unused_variables)]
#fn main() {
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
}
#}

带有FuseFuturesUnorderedselect循环中的并发任务

有个不太好找但是很趁手的函数叫Fuse::terminated()。这个函数允许构造已经被终止的空 future,并且能够在之后填进需要运行的future。 这个在一个任务需要select循环中运行但是它本身是在select循环中创建的场景中很好用。

注意.select_next_some()函数的是使用。这可以用在select上,并且只运行从stream返回的 Some(_)值而忽略None


# #![allow(unused_variables)]
#fn main() {
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.
                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.set(run_on_new_num(new_num).fuse());
            },
            // Run the `run_on_new_num_fut`
            () = run_on_new_num_fut => {},
            // panic if everything completed, since the `interval_timer` should
            // keep yielding values indefinitely.
            complete => panic!("`interval_timer` completed unexpectedly"),
        }
    }
}
#}

当有很多份相同future的拷贝同时执行时,使用FutureUnordered类型。下面的例子和上面的 例子很类似,但会运行run_on_new_num_fut的所有拷贝都到完成状态,而不是当一个新拷贝 创建时就中断他们。它也会打印run_on_new_num_fut的返回值:


# #![allow(unused_variables)]
#fn main() {
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 }

// Runs `run_on_new_num` with the latest number
// retrieved from `get_new_num`.
//
// `get_new_num` is re-run every time a timer elapses,
// immediately cancelling the currently running
// `run_on_new_num` and replacing it with the newly
// returned value.
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.
                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_futs.push(run_on_new_num(new_num));
            },
            // Run the `run_on_new_num_futs` and check if any have completed
            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.
            complete => panic!("`interval_timer` completed unexpectedly"),
        }
    }
}

#}