Rusoto Async с использованием комбинатора FuturesOrdered - PullRequest
Новая
       37

Rusoto Async с использованием комбинатора FuturesOrdered

0 голосов
/ 07 мая 2019

Я пытаюсь отправить параллельные асинхронные запросы Rusoto SQS, используя FuturesOrdered: 

use futures::prelude::*; // 0.1.26
use futures::stream::futures_unordered::FuturesUnordered;
use rusoto_core::{Region, HttpClient}; // 0.38.0
use rusoto_credential::EnvironmentProvider; // 0.17.0
use rusoto_sqs::{SendMessageBatchRequest, SendMessageBatchRequestEntry, Sqs, SqsClient}; // 0.38.0

fn main() {
    let client = SqsClient::new_with(
        HttpClient::new().unwrap(),
        EnvironmentProvider::default(),
        Region::UsWest2,
    );

    let messages: Vec<u32> = (1..12).map(|n| n).collect();
    let chunks: Vec<_> = messages.chunks(10).collect();

    let tasks: FuturesUnordered<_> = chunks.into_iter().map(|c| {
        let batch = create_batch(c);
        client.send_message_batch(batch)
    }).collect();

    let tasks = tasks
        .for_each(|t| {
            println!("{:?}", t);
            Ok(())
        })
        .map_err(|e| println!("{}", e));

    tokio::run(tasks);
}

fn create_batch(ids: &[u32]) -> SendMessageBatchRequest {
    let queue_url = "https://sqs.us-west-2.amazonaws.com/xxx/xxx".to_string();
    let entries = ids
        .iter()
        .map(|id| SendMessageBatchRequestEntry {
            id: id.to_string(),
            message_body: id.to_string(),
            ..Default::default()
        })
        .collect();

    SendMessageBatchRequest {
        entries,
        queue_url,
    }
}

Задачи завершаются правильно, но tokio::run(tasks) не останавливается.Я предполагаю, что это из-за tasks.for_each() заставит его продолжать работать и искать больше фьючерсов?

Почему tokio::run(tasks) не останавливается?Правильно ли я использую FuturesOrdered?

Я также немного беспокоюсь об использовании памяти при создании до 60 000 фьючерсов для завершения и вставке их в комбинатор FuturesUnordered.

1 Ответ

0 голосов
/ 12 мая 2019

Я обнаружил, что это была SqsClient в основной функции, которая заставляла его блокироваться, поскольку он все еще выполняет некоторую домашнюю работу, даже если задачи завершены.

Решение, предоставленное одним из людей Rusoto, состояло в том, чтобы добавить это чуть выше tokio::run 

std::mem::drop(client);
...