rust - 将新任务添加到 tokio 事件循环并在失败时重试任务
问题描述
我正在尝试编写一个可以从同一服务器执行获取请求的 tokio 事件循环,具有以下特征:
- 应该使用连接池
- get请求一般比较慢(>1s),所以需要并行执行
- 服务器可能没有响应,所以我需要超时。如果未收到请求,请再次发送
- 轮询接收器以获取必须下载的新 URL。它们应该被添加到事件循环中
到目前为止,在我的尝试中,我已经设法让这 4 个项目的不同组合起作用,但从来没有一起工作。我的主要问题是我不太明白如何向 tokio 事件循环添加新的期货。
我假设我需要使用loop_fn
轮询接收器的主循环并handle.spawn
生成新任务?handle.spawn
只允许未来Result<(),()>
,所以我不能使用它的输出在失败时重新生成作业,所以我需要将重试检查移到那个未来?
以下是批量接受和处理 url 的尝试(因此没有连续轮询),并且有超时(但没有重试):
fn place_dls(&mut self, reqs: Vec<String>) {
let mut core = Core::new().unwrap();
let handle = core.handle();
let timeout = Timeout::new(Duration::from_millis(5000), &handle).unwrap();
let send_dls = stream::iter_ok::<_, reqwest::Error>(reqs.iter().map(|o| {
// send with request through an async reqwest client in self
}));
let rec_dls = send_dls.buffer_unordered(dls.len()).for_each(|n| {
n.into_body().concat2().and_then(|full_body| {
debug!("Received: {:#?}", full_body);
// TODO: how to put the download back in the queue if failure code is received?
})
});
let work = rec_dls.select2(timeout).then(|res| match res {
Ok(Either::A((got, _timeout))) => {
Ok(got)
},
Ok(Either::B((_timeout_error, _get))) => {
// TODO: put back in queue
Err(io::Error::new(
io::ErrorKind::TimedOut,
"Client timed out while connecting",
).into())
}
Err(Either::A((get_error, _timeout))) => Err(get_error.into()),
Err(Either::B((timeout_error, _get))) => Err(timeout_error.into()),
});
core.run(work);
}
loop_fn
可悲的是,我的尝试失败了。
解决方案
我假设我需要在主循环中使用 loop_fn
我建议稍微另一种方法:实现futures::sync::mpsc::Receiver
流消费者而不是循环。
它可以被视为某种主进程:在通过Receiver
tokio 任务接收到 url 后,可以生成内容下载。Sender
那么重试就没有问题了:只需通过其端点再次将失败或超时的url发送到主通道。
这是一个工作代码草图:
extern crate futures;
extern crate tokio;
use std::{io, time::{Duration, Instant}};
use futures::{
Sink,
Stream,
stream,
sync::mpsc,
future::Future,
};
use tokio::{
runtime::Runtime,
timer::{Delay, Timeout},
};
fn main() -> Result<(), io::Error> {
let mut rt = Runtime::new()?;
let executor = rt.executor();
let (tx, rx) = mpsc::channel(0);
let master_tx = tx.clone();
let master = rx.for_each(move |url: String| {
let download_future = download(&url)
.map(|_download_contents| {
// TODO: actually do smth with contents
()
});
let timeout_future =
Timeout::new(download_future, Duration::from_millis(2000));
let job_tx = master_tx.clone();
let task = timeout_future
.or_else(move |error| {
// actually download error or timeout, retry
println!("retrying {} because of {:?}", url, error);
job_tx.send(url).map(|_| ()).map_err(|_| ())
});
executor.spawn(task);
Ok(())
});
rt.spawn(master);
let urls = vec![
"http://url1".to_string(),
"http://url2".to_string(),
"http://url3".to_string(),
];
rt.executor()
.spawn(tx.send_all(stream::iter_ok(urls)).map(|_| ()).map_err(|_| ()));
rt.shutdown_on_idle().wait()
.map_err(|()| io::Error::new(io::ErrorKind::Other, "shutdown failure"))
}
#[derive(Debug)]
struct DownloadContents;
#[derive(Debug)]
struct DownloadError;
fn download(url: &str) -> Box<Future<Item = DownloadContents, Error = DownloadError> + Send> {
// TODO: actually download here
match url {
// url2 always fails
"http://url2" => {
println!("FAILED downloading: {}", url);
let future = Delay::new(Instant::now() + Duration::from_millis(1000))
.map_err(|_| DownloadError)
.and_then(|()| Err(DownloadError));
Box::new(future)
},
// url3 always timeouts
"http://url3" => {
println!("TIMEOUT downloading: {}", url);
let future = Delay::new(Instant::now() + Duration::from_millis(5000))
.map_err(|_| DownloadError)
.and_then(|()| Ok(DownloadContents));
Box::new(future)
},
// everything else succeeds
_ => {
println!("SUCCESS downloading: {}", url);
let future = Delay::new(Instant::now() + Duration::from_millis(50))
.map_err(|_| DownloadError)
.and_then(|()| Ok(DownloadContents));
Box::new(future)
},
}
}
推荐阅读
- android - 如何测量对 findViewByID 的额外调用 - RecyclerView
- http - 关于 ::http::geturl 令牌的使用
- django - 如何使用基于函数的视图删除对象?
- hadoop - 为什么 Hbase 不使用集群中的所有节点?
- julia - 具有多变量功能输入的一维集成
- javascript - array.includes 返回 false ,其中搜索的名称在数组中
- powershell - 我想在 Jenkins 中执行我的 powershell 脚本并显示结果 OK/NOK
- c# - Unity 容器中的方法重载
- r - 当有 3 列时,R 中的插值
- ios - 如何创建一个接受 SCNNodes 的自定义 ScrollView