我有一个
main
函数,我在其中创建了一个 Tokio 运行时并在其上运行两个 futures。
use tokio;
fn main() {
let mut runtime = tokio::runtime::Runtime::new().unwrap();
runtime.spawn(MyMegaFutureNumberOne {});
runtime.spawn(MyMegaFutureNumberTwo {});
// Some code to 'join' them after receiving an OS signal
}
如何收到
SIGTERM
,等待所有未完成的任务(NotReady
s)并退出应用程序?
处理信号很棘手,解释如何处理所有可能的情况太宽泛了。信号的实现不是跨平台标准的,所以我的回答是特定于 Linux 的。如果想更跨平台,使用POSIX函数
sigaction
结合pause
;这将为您提供更多控制权。
tokio 的文档有一个很棒的 getting started guide for signal in tokio。因此,我会尝试添加我自己的建议。
我的一般建议是有一个任务来为我们处理信号,然后你在你的其他任务中使用一个 watch channel,如果 watch channel 状态改变,它将停止。
我的第二个建议是将
biased
与等待您的期货的 select
一起使用,这很重要,因为您通常想知道是否立即收到信号并且之前没有做其他事情。这可能是一个经常准备好的繁忙循环的问题,你永远不会得到你的信号未来分支。请仔细阅读documentation关于biased
.
use core::time::Duration;
use tokio::{
select,
signal::unix::{signal, SignalKind},
sync::watch,
time::sleep,
};
#[tokio::main]
async fn main() {
let (stop_tx, mut stop_rx) = watch::channel(());
tokio::spawn(async move {
let mut sigterm = signal(SignalKind::terminate()).unwrap();
let mut sigint = signal(SignalKind::interrupt()).unwrap();
loop {
select! {
_ = sigterm.recv() => println!("Recieve SIGTERM"),
_ = sigint.recv() => println!("Recieve SIGTERM"),
};
stop_tx.send(()).unwrap();
}
});
loop {
select! {
biased;
_ = stop_rx.changed() => break,
i = some_operation(42) => {
println!("Result is {i}");
unsafe { libc::raise(libc::SIGTERM)};
},
}
}
}
async fn some_operation(i: u64) -> u64 {
println!("Task started.");
sleep(Duration::from_millis(i)).await;
println!("Task shutting down.");
i
}
您可以根据需要克隆频道的接收器,这将有效地处理信号。
实现你想要的一种方法是使用 tokio_signal 箱子来捕捉信号,像这样:(文档示例)
extern crate futures;
extern crate tokio;
extern crate tokio_signal;
use futures::prelude::*;
use futures::Stream;
use std::time::{Duration, Instant};
use tokio_signal::unix::{Signal, SIGINT, SIGTERM};
fn main() -> Result<(), Box<::std::error::Error>> {
let mut runtime = tokio::runtime::Runtime::new()?;
let sigint = Signal::new(SIGINT).flatten_stream();
let sigterm = Signal::new(SIGTERM).flatten_stream();
let stream = sigint.select(sigterm);
let deadline = tokio::timer::Delay::new(Instant::now() + Duration::from_secs(5))
.map(|()| println!("5 seconds are over"))
.map_err(|e| eprintln!("Failed to wait: {}", e));
runtime.spawn(deadline);
let (item, _rest) = runtime
.block_on_all(stream.into_future())
.map_err(|_| "failed to wait for signals")?;
let item = item.ok_or("received no signal")?;
if item == SIGINT {
println!("received SIGINT");
} else {
assert_eq!(item, SIGTERM);
println!("received SIGTERM");
}
Ok(())
}
此程序将等待所有当前任务完成并捕获选定的信号。这似乎在 Windows 上不起作用,因为它会立即关闭程序。
对于 Tokio 版本 1.x.y,Tokio 官方教程有一个关于这个主题的页面:Graceful shutdown