如何解决如何在循环中生成异步方法?
我有一个对象向量,这些对象具有一个resolve()
方法,该方法使用reqwest
查询外部Web API。在每个对象上调用resolve()
方法之后,我想打印每个请求的结果。
这是我的半异步代码,可以编译和运行(但不是真正异步):
for mut item in items {
item.resolve().await;
item.print_result();
}
我尝试使用tokio::join!
产生所有异步调用并等待它们完成,但是我可能做错了事:
tokio::join!(items.iter_mut().for_each(|item| item.resolve()));
这是我遇到的错误:
error[E0308]: mismatched types
--> src\main.rs:25:51
|
25 | tokio::join!(items.iter_mut().for_each(|item| item.resolve()));
| ^^^^^^^^^^^^^^ expected `()`,found opaque type
|
::: src\redirect_definition.rs:32:37
|
32 | pub async fn resolve(&mut self) {
| - the `Output` of this `async fn`'s found opaque type
|
= note: expected unit type `()`
found opaque type `impl std::future::Future`
如何为所有实例一次调用resolve()
方法?
这段代码反映了答案-现在我正在处理我不太了解的借位检查器错误-我应该用'static
注释一些变量吗?
let mut items = get_from_csv(path);
let tasks: Vec<_> = items
.iter_mut()
.map(|item| tokio::spawn(item.resolve()))
.collect();
for task in tasks {
task.await;
}
for item in items {
item.print_result();
}
error[E0597]: `items` does not live long enough
--> src\main.rs:18:25
|
18 | let tasks: Vec<_> = items
| -^^^^
| |
| _________________________borrowed value does not live long enough
| |
19 | | .iter_mut()
| |___________________- argument requires that `items` is borrowed for `'static`
...
31 | }
| - `items` dropped here while still borrowed
error[E0505]: cannot move out of `items` because it is borrowed
--> src\main.rs:27:17
|
18 | let tasks: Vec<_> = items
| -----
| |
| _________________________borrow of `items` occurs here
| |
19 | | .iter_mut()
| |___________________- argument requires that `items` is borrowed for `'static`
...
27 | for item in items {
| ^^^^^ move out of `items` occurs here
解决方法
由于您希望并行等待期货,因此可以spawn将其分成并行运行的各个任务。由于它们彼此独立运行且不会与生成它们的线程独立运行,因此您可以按任何顺序等待它们的句柄。
理想情况下,您会这样写:
// spawn tasks that run in parallel
let tasks: Vec<_> = items
.iter_mut()
.map(|item| tokio::spawn(item.resolve()))
.collect();
// now await them to get the resolve's to complete
for task in tasks {
task.await.unwrap();
}
// and we're done
for item in &items {
item.print_result();
}
但是借阅检查器将拒绝此操作,因为item.resolve()
返回的Future拥有借用的对item
的引用。该引用将传递到tokio::spawn()
,后者将其传递给另一个线程,并且编译器无法证明item
会超过该线程。 (当您想send reference to local data to a thread时会遇到相同的问题。)
对此有几种可能的解决方案;我发现最优雅的方法是将项目移动到传递给tokio::spawn()
的异步闭包中,并让任务完成后将其交还给您。基本上,您使用items
向量创建任务并立即从等待的结果中重新构造它:
// note the use of `into_iter()` to consume `items`
let tasks: Vec<_> = items
.into_iter()
.map(|mut item| {
tokio::spawn(async {
item.resolve().await;
item
})
})
.collect();
// await the tasks for resolve's to complete and give back our items
let mut items = vec![];
for task in tasks {
items.push(task.await.unwrap());
}
// verify that we've got the results
for item in &items {
item.print_result();
}
playground中的可运行代码。
请注意,futures
板条箱包含一个join_all
函数,该函数与您需要的函数相似,不同之处在于它将轮询各个期货,而不确保它们并行运行。我们可以编写使用join_parallel
的通用join_all
,但也可以使用tokio::spawn
来并行执行:
async fn join_parallel<T: Send + 'static>(
futs: impl IntoIterator<Item = impl Future<Output = T> + Send + 'static>,) -> Vec<T> {
let tasks: Vec<_> = futs.into_iter().map(tokio::spawn).collect();
// unwrap the Result because it is introduced by tokio::spawn()
// and isn't something our caller can handle
futures::future::join_all(tasks)
.await
.into_iter()
.map(Result::unwrap)
.collect()
}
使用此功能,回答问题所需的代码简化为:
let items = join_parallel(items.into_iter().map(|mut item| async {
item.resolve().await;
item
})).await;
for item in &items {
item.print_result();
}
再次在playground中运行代码。
版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 dio@foxmail.com 举报,一经查实,本站将立刻删除。