Rust异步信号处理库signal-hook-async-std的使用指南

介绍

signal-hook-async-std是一个Rust库，它为async-std运行时提供了异步信号处理能力。这个库建立在signal-hook之上，专门为async-std生态系统设计，允许开发者在异步上下文中优雅地处理Unix信号。

主要特点：

• 跨平台支持（主要针对Unix-like系统）
• 与async-std运行时无缝集成
• 提供类似事件驱动的信号处理方式
• 线程安全且高效

安装

在Cargo.toml中添加依赖：

[dependencies]
signal-hook-async-std = "0.2"
async-std = "1.12"

基本使用方法

监听单个信号

use async_std::task;
use signal_hook_async_std::Signals;

async fn handle_signals() {
    let signals = Signals::new(&[signal_hook::consts::SIGINT]).unwrap();
    
    while let Some(signal) = signals.next().await {
        match signal {
            signal_hook::consts::SIGINT => {
                println!("Received SIGINT (Ctrl+C), shutting down...");
                break;
            }
            _ => unreachable!(),
        }
    }
}

fn main() {
    task::block_on(handle_signals());
}

监听多个信号

use async_std::task;
use signal_hock_async_std::Signals;
use signal_hook::consts::{SIGINT, SIGTERM, SIGQUIT};

async fn handle_multiple_signals() {
    let signals = Signals::new(&[SIGINT, SIGTERM, SIGQUIT]).unwrap();
    
    while let Some(signal) = signals.next().await {
        match signal {
            SIGINT => println!("Received SIGINT (Ctrl+C)"),
            SIGTERM => println!("Received SIGTERM (termination request)"),
            SIGQUIT => println!("Received SIGQUIT (Ctrl+\\)"),
            _ => unreachable!(),
        }
    }
}

fn main() {
    task::block_on(handle_multiple_signals());
}

高级用法

结合async-std任务使用

use async_std::{task, prelude::*};
use signal_hook_async_std::Signals;
use signal_hook::consts::SIGINT;

async fn worker_task() {
    for i in 1..=10 {
        task::sleep(std::time::Duration::from_secs(1)).await;
        println!("Working... {}", i);
    }
}

async fn signal_handler() {
    let signals = Signals::new(&[SIGINT]).unwrap();
    signals.forever().next().await;
    println!("Signal received, stopping gracefully...");
}

#[async_std::main]
async fn main() {
    let worker = task::spawn(worker_task());
    let signal = task::spawn(signal_handler());
    
    worker.race(signal).await;
}

自定义信号处理逻辑

use async_std::{task, sync::Arc};
use signal_hook_async_std::Signals;
use signal_hook::consts::SIGUSR1;
use std::sync::atomic::{AtomicUsize, Ordering};

async fn handle_custom_signal(counter: Arc<AtomicUsize>) {
    let signals = Signals::new(&[SIGUSR1]).unwrap();
    
    while let Some(_) = signals.next().await {
        let current = counter.fetch_add(1, Ordering::SeqCst);
        println!("Received SIGUSR1 signal (count: {})", current + 1);
    }
}

#[async_std::main]
async fn main() {
    let counter = Arc::new(AtomicUsize::new(0));
    let handler = handle_custom_signal(counter.clone());
    
    println!("Send SIGUSR1 to process {} to increment counter", std::process::id());
    handler.await;
}

注意事项

1. 这个库主要适用于Unix-like系统，Windows支持有限
2. 某些信号（如SIGKILL）不能被捕获或处理
3. 在异步上下文中处理信号时，应尽量保持处理逻辑简短
4. 多个任务监听相同信号时，只有一个会收到通知

实际应用示例：优雅关闭服务

use async_std::{task, net::{TcpListener, TcpStream}, prelude::*};
use signal_hook_async_std::Signals;
use signal_hook::consts::{SIGINT, SIGTERM};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;

async fn handle_connection(mut stream: TcpStream, running: Arc<AtomicBool>) {
    let mut buf = vec![0u8; 1024];
    while running.load(Ordering::Relaxed) {
        match stream.read(&mut buf).await {
            Ok(n) if n == 0 => break,
            Ok(n) => {
                println!("Received {} bytes", n);
                if let Err(e) = stream.write_all(&buf[..n].await {
                    eprintln!("Write error: {}", e);
                    break;
                }
            }
            Err(e) => {
                eprintln!("Read error: {}", e);
                break;
            }
        }
    }
}

async fn run_server(running: Arc<AtomicBool>) -> std::io::Result<()> {
    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    println!("Server listening on port 8080");
    
    while running.load(Ordering::Relaxed) {
        match listener.accept().await {
            Ok((stream, _)) => {
                let running_clone = running.clone();
                task::spawn(handle_connection(stream, running_clone));
            }
            Err(e) => eprintln!("Connection failed: {}", e),
        }
    }
    
    println!("Server shutting down gracefully...");
    Ok(())
}

#[async_std::main]
async fn main() -> std::io::Result<()> {
    let running = Arc::new(AtomicBool::new(true));
    let running_clone = running.clone();
    
    let signals = Signals::new(&[SIGINT, SIGTERM]).unwrap();
    let signal_handler = async move {
        signals.forever().next().await;
        running_clone.store(false, Ordering::Relaxed);
    };
    
    let server = run_server(running);
    server.race(signal_handler).await?;
    
    Ok(())
}

这个示例展示了如何结合signal-hook-async-std和async-std创建一个可以优雅关闭的TCP服务器，在收到终止信号时能够干净地关闭所有连接。