Bir döngüde eşzamansız yöntemleri nasıl oluşturabilirim?

Question

Harici bir web API'sini sorgulamak için resolve()kullanan bir yöntemi olan bir nesne vektörüm var reqwest. resolve()Her nesnede yöntemi çağırdıktan sonra , her isteğin sonucunu yazdırmak istiyorum.

İşte derleyen ve çalışan (ama gerçekten eşzamansız olmayan) yarı eşzamansız kodum:

for mut item in items {
    item.resolve().await;

    item.print_result();
}

tokio::join!Tüm eşzamansız çağrıları oluşturmayı ve bitmelerini beklemeyi denedim , ancak muhtemelen yanlış bir şey yapıyorum:

tokio::join!(items.iter_mut().for_each(|item| item.resolve()));

İşte aldığım hata:

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()Tüm örnekler için yöntemleri aynı anda nasıl çağırabilirim ?

---

Bu kod cevabı yansıtıyor - şimdi gerçekten anlamadığım borçlanma denetleyicisi hatalarıyla uğraşıyorum - bazı değişkenlerime açıklama eklemeli miyim '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

Answer 1

Eğer paralel gelecekleri üzerinde bekliyor istediğimizden yapabilirsiniz yumurtlamaya paralel olarak çalışan bireysel görevlerin içine. Birbirlerinden ve onları oluşturan iplikten bağımsız olarak hareket ettikleri için, kollarını istediğiniz sırada bekleyebilirsiniz.

İdeal olarak şöyle bir şey yazarsın:

// 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();
}

Ancak bu, ödünç alma denetçisi tarafından reddedilecektir, çünkü geri dönen gelecek item.resolve(), ödünç alınmış bir referansı barındırmaktadır item. Referans tokio::spawn()başka bir iş parçacığına iletilir ve derleyici bunun itembu iş parçacığını aşacağını kanıtlayamaz . ( Bir iş parçacığına yerel verilere başvuru göndermek istediğinizde de aynı tür bir sorunla karşılaşılır .)

Bunun birkaç olası çözümü vardır; En zarif bulduğum şey, öğeleri geçilen zaman uyumsuz kapağa taşımaktokio::spawn() ve görev bittiğinde bunları size geri vermek. Temel olarak items, görevleri oluşturmak için vektörü kullanırsınız ve beklenen sonuçlardan hemen yeniden yapılandırırsınız:

// 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();
}

Oyun alanında çalıştırılabilir kod .

futuresSandığın join_allihtiyacınız olana benzer bir işlev içerdiğine dikkat edin , ancak tek tek vadeli işlemleri paralel olarak çalıştıklarından emin olmadan yoklar. Biz genel yazabilir join_parallelkullandığı bu join_alldeğil, aynı zamanda kullandığı tokio::spawnparalel yürütülmesine almak için:

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()
}

Bu işlevi kullanarak soruyu yanıtlamak için gereken kod şu şekilde özetlenebilir:

let items = join_parallel(items.into_iter().map(|mut item| async {
    item.resolve().await;
    item
})).await;
for item in &items {
    item.print_result();
}

Yine, oyun alanında çalıştırılabilir kod .