招数单一，杀伤力惊人，说的就是 .await
，但是光用它，还真做不到一招鲜吃遍天。比如我们该如何同时运行多个任务，而不是使用
.await 慢悠悠地排队完成。
futures 包中提供了很多实用的工具，其中一个就是
join! 宏， 它允许我们同时等待多个不同 Future
的完成，且可以并发地运行这些 Future。
先来看一个不是很给力的、使用.await的版本:
1
2
3
4
5
| async fn enjoy_book_and_music() -> (Book, Music) {
let book = enjoy_book().await;
let music = enjoy_music().await;
(book, music)
}
|
这段代码可以顺利运行，但是有一个很大的问题，就是必须先看完书后，才能听音乐。咱们以前，谁又不是那个摇头晃脑爱读书(耳朵里偷偷塞着耳机，听的正
high)的好学生呢？
要支持同时看书和听歌，有些人可能会凭空生成下面代码：
1
2
3
4
5
6
|
async fn enjoy_book_and_music() -> (Book, Music) {
let book_future = enjoy_book();
let music_future = enjoy_music();
(book_future.await, music_future.await)
}
|
看上去像模像样，嗯，在某些语言中也许可以，但是 Rust
不行。因为在某些语言中，Future一旦创建就开始运行，等到返回的时候，基本就可以同时结束并返回了。
但是 Rust 中的 Future 是惰性的，直到调用
.await 时，才会开始运行。而那两个 await
由于在代码中有先后顺序，因此它们是顺序运行的。
为了正确的并发运行两个 Future ， 我们来试试
futures::join! 宏:
1
2
3
4
5
6
7
| use futures::join;
async fn enjoy_book_and_music() -> (Book, Music) {
let book_fut = enjoy_book();
let music_fut = enjoy_music();
join!(book_fut, music_fut)
}
|
Duang，目标顺利达成。同时 join!
会返回一个元组，里面的值是对应的 Future
执行结束后输出的值。
如果希望同时运行一个数组里的多个异步任务，可以使用
futures::future::join_all 方法
由于 join! 必须等待它管理的所有 Future
完成后才能完成，如果你希望在某一个 Future
报错后就立即停止所有 Future 的执行，可以使用
try_join!，特别是当 Future 返回
Result 时:
1
2
3
4
5
6
7
8
9
10
| 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 方法将错误进行转换:
1
2
3
4
5
6
7
8
9
10
11
12
13
| 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)
}
|
join! 很好很强大，但是人无完人，J 无完 J,
它有一个很大的问题。
join! 只有等所有 Future
结束后，才能集中处理结果，如果你想同时等待多个 Future
，且任何一个 Future 结束后，都可以立即被处理，可以考虑使用
futures::select!:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
| use futures::{
future::FutureExt,
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!("任务1率先完成"),
() = t2 => println!("任务2率先完成"),
}
}
|
上面的代码会同时并发地运行 t1 和 t2，
无论两者哪个先完成，都会调用对应的 println!
打印相应的输出，然后函数结束且不会等待另一个任务的完成。
但是，在实际项目中，我们往往需要等待多个任务都完成后，再结束，像上面这种其中一个任务结束就立刻结束的场景着实不多。
select!还支持 default 和
complete 分支:
complete 分支当所有的 Future 和
Stream 完成后才会被执行，它往往配合 loop
使用，loop 用于循环完成所有的 Futuredefault 分支，若没有任何 Future 或
Stream 处于 Ready 状态，
则该分支会被立即执行
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| use futures::future;
use futures::select;
pub fn main() {
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 => panic!(),
};
}
assert_eq!(total, 10);
}
|
以上代码 default 分支由于最后一个运行，而在它之前
complete 分支已经通过 break 跳出了循环，因此
default 永远不会被执行。
如果你希望 default 也有机会露下脸，可以将
complete 的 break 修改为其它的，例如
println!("completed!")，然后再观察下运行结果。
再回到 select 的第一个例子中，里面有一段代码长这样：
1
2
3
4
| let t1 = task_one().fuse();
let t2 = task_two().fuse();
pin_mut!(t1, t2);
|
当时没有展开讲，相信大家也有疑惑，下面我们来一起看看。
首先，.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:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
| 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
}
|
一个很实用但又鲜为人知的函数是 Fuse::terminated()
，可以使用它构建一个空的 Future
，空自然没啥用，但是如果它能在后面再被填充呢？
考虑以下场景：当你要在 select
循环中运行一个任务，但是该任务却是在 select
循环内部创建时，上面的函数就非常好用了。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
| 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() => {
if get_new_num_fut.is_terminated() {
get_new_num_fut.set(get_new_num().fuse());
}
},
new_num = get_new_num_fut => {
run_on_new_num_fut.set(run_on_new_num(new_num).fuse());
},
() = run_on_new_num_fut => {},
complete => panic!("`interval_timer` completed unexpectedly"),
}
}
}
|
当某个 Future 有多个拷贝都需要同时运行时，可以使用
FuturesUnordered
类型。下面的例子跟上个例子大体相似，但是它会将
run_on_new_num_fut
的每一个拷贝都运行到完成，而不是像之前那样一旦创建新的就终止旧的。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
| 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 }
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() => {
if get_new_num_fut.is_terminated() {
get_new_num_fut.set(get_new_num().fuse());
}
},
new_num = get_new_num_fut => {
run_on_new_num_futs.push(run_on_new_num(new_num));
},
res = run_on_new_num_futs.select_next_some() => {
println!("run_on_new_num_fut returned {:?}", res);
},
complete => panic!("`interval_timer` completed unexpectedly"),
}
}
}
|