Rusoto 异步使用 FuturesOrdered 组合器
Rusoto async using FuturesOrdered combinator
我正在尝试使用 FuturesOrdered
:
发送并行异步 Rusoto SQS 请求
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 个 futures 以完成并将它们推入 FuturesUnordered
组合器时的内存使用情况。
我发现是 main 函数中的 SqsClient
导致它阻塞,因为即使任务已完成,它仍在做一些家务。
一位 Rusoto 人提供的解决方案是将其添加到 tokio::run
正上方
std::mem::drop(client);
我正在尝试使用 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 个 futures 以完成并将它们推入 FuturesUnordered
组合器时的内存使用情况。
我发现是 main 函数中的 SqsClient
导致它阻塞,因为即使任务已完成,它仍在做一些家务。
一位 Rusoto 人提供的解决方案是将其添加到 tokio::run
std::mem::drop(client);