在Rust中等待未来任务列表

Question

我尝试创建一个能够持续发现新工作并为这些工作项维护一组未来数据的未来系统。我想确保我的主要未来任务不会被长时间阻塞，并使我的工作能够并发执行。

以下是我正在尝试实现的大致概述。具体而言，“isDone”不存在，根据文档理解，似乎也不是使用Rust中未来的有效方式。有没有其它常用的方法来实现这种操作呢？

use std::collections::HashMap;
use tokio::runtime::Runtime;

async fn find_work() -> HashMap<i64, String> {
    // Go read from the DB or something...
    let mut work = HashMap::new();
    work.insert(1, "test".to_string());
    work.insert(2, "test".to_string());
    return work;
}

async fn do_work(id: i64, value: String) -> () {
    // Result<(), Error> {
    println!("{}: {}", id, value);
}

async fn async_main() -> () {
    let mut pending_work = HashMap::new();
    loop {
        for (id, value) in find_work().await {
            if !pending_work.contains_key(&id) {
                let fut = do_work(id, value);
                pending_work.insert(id, fut);
            }
        }

        pending_work.retain(|id, fut| {
            if isDone(fut) {
                // do something with the result
                false
            } else {
                true
            }
        });
    }
}

fn main() {
    let runtime = Runtime::new().unwrap();
    let exec = runtime.executor();

    exec.spawn(async_main());
    runtime.shutdown_on_idle();
}

- Nick

---

也许你正在寻找类似于 join!() 或者 select!() 的东西。 - Svetlin Zarev

谢谢建议，我看了一下但不幸的是join！会阻塞主循环，而我的主循环需要检查新任务，而且select和join都只接受固定数量的futures而不是map或vec，所以使用起来感觉有些笨拙。我会尝试使用select，看看是否运行良好。 - Nick

Futures只有在被轮询时才会执行工作，所以也许你应该只是poll()它们？如果future已经完成，你将得到Poll::Ready，否则- Poll::Pending，这与你的isDone()方法相同，除了它返回一个具有两个变体的枚举，而不是true/false。 - Svetlin Zarev

虽然不完全符合描述，但我最终使用了 futures::future::join_all（https://docs.rs/futures/0.3.5/futures/future/fn.join_all.html），它可以接受任何 IntoInterator，并且对我来说效果很好。 - Nick

@Nick 把那个作为答案发布。 - Dull Bananas

Answer 1

正如 @Nick 已经回答的那样，您可以使用 futures::future::join_all 来自 futures crate。

- Lukas