Tokio 任务 Panic 隔离与恢复

示例说明

spawn 出的任务 panic 不会直接打崩进程——异常被捕获并封存进 JoinError，await 句柄时才浮出水面。is_panic 区分两种失败：恐慌是代码缺陷或脏数据，取消是正常协同；into_panic 把载荷还原为具体类型，恐慌消息因此可进入告警而不是只有堆栈。重启策略要配上限——恐慌往往是系统性的，无限重启等于忙循环。

技术要点

• JoinError 封存 spawn 任务的 panic
• into_panic 还原具体载荷类型
• 重启带上限防恐慌风暴

代码解读

supervised_batch 收集五个任务的句柄逐个 await：二号任务恐慌，其载荷被向下转型为 String 拼进失败清单，其余结果正常收集。RestartingWorker 的循环里恐慌触发重启计数、超过两次放弃；闭包的前两次调用恐慌、第三次成功，验证自愈闭环。

使用建议

生产代码优先修复恐慌根因，恢复只是止损。恐慌任务持有的锁要保证释放语义。兜底告警接入重启计数而不是零星恐慌。

代码示例

use tokio::task::JoinHandle;

async fn risky_compute(index: usize) -> usize {
    if index == 2 {
        panic!("worker {index} hit unrecoverable state");
    }
    index * 100
}

async fn supervised_batch() -> (Vec<usize>, Vec<String>) {
    let handles: Vec<JoinHandle<usize>> = (0..5)
        .map(|i| tokio::spawn(risky_compute(i)))
        .collect();

    let mut results = Vec::new();
    let mut failures = Vec::new();

    for (position, handle) in handles.into_iter().enumerate() {
        match handle.await {
            Ok(value) => results.push(value),
            Err(join_err) if join_err.is_panic() => {
                let message = join_err
                    .into_panic()
                    .downcast_ref::<String>()
                    .cloned()
                    .unwrap_or_else(|| "unknown panic payload".into());
                failures.push(format!("slot {position}: {message}"));
            }
            Err(join_err) => {
                failures.push(format!("slot {position}: {join_err}"));
            }
        }
    }
    (results, failures)
}

struct RestartingWorker {
    max_restarts: usize,
}

impl RestartingWorker {
    async fn run_with_restart<F, Fut>(&self, mut job: F)
    where
        F: FnMut() -> Fut,
        Fut: std::future::Future<Output = usize>,
    {
        let mut restarts = 0;
        loop {
            let handle = tokio::spawn(job());
            match handle.await {
                Ok(count) => {
                    println!("worker completed {count} items");
                    return;
                }
                Err(err) if err.is_panic() => {
                    restarts += 1;
                    if restarts > self.max_restarts {
                        println!("giving up after {restarts} restarts");
                        return;
                    }
                    println!("worker panicked, restart {restarts}");
                }
                Err(_) => return,
            }
        }
    }
}

#[tokio::main]
async fn main() {
    let (results, failures) = supervised_batch().await;
    println!("results: {results:?}");
    println!("failures: {failures:?}");

    let worker = RestartingWorker { max_restarts: 2 };
    let mut invocations = 0;
    worker
        .run_with_restart(move || {
            invocations += 1;
            let attempt = invocations;
            async move {
                if attempt < 3 {
                    panic!("transient failure {attempt}");
                }
                42
            }
        })
        .await;
}