如何实现带超时的Tokio异步任务？REST API调用超时终止需求

核心技术关键词 #

• tokio::time::timeout：为任务设置超时阈值
• JoinHandle::abort()：主动终止Tokio生成的异步任务
• tokio_util::sync::CancellationToken：灵活的多任务联动取消机制
• 可取消await点：确保任务内部能响应取消信号

代码示例 #

方案1：timeout + JoinHandle::abort() 单任务超时中断 #

适合简单场景，给单个任务独立设置超时，超时后主动终止：

use tokio::time::{timeout, Duration};
use tokio::spawn;
use std::error::Error;

async fn call_rest_api(api_url: &str) -> Result<String, Box<dyn Error>> {
    // 模拟REST API调用，确保内部await操作支持取消
    let resp = reqwest::get(api_url).await?;
    Ok(resp.text().await?)
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    // 启动第一个API任务并获取句柄
    let task1_handle = spawn(call_rest_api("https://api.example.com/service1"));
    // 设置5秒超时
    match timeout(Duration::from_secs(5), task1_handle).await {
        Ok(Ok(result)) => println!("任务1完成: {}", result),
        Ok(Err(e)) => eprintln!("任务1执行出错: {}", e),
        Err(_) => {
            eprintln!("任务1超时，已终止");
            // 调用abort确保任务彻底停止调度
            task1_handle.abort();
        }
    }

    // 同理处理第二个任务
    let task2_handle = spawn(call_rest_api("https://api.example.com/service2"));
    match timeout(Duration::from_secs(3), task2_handle).await {
        Ok(Ok(result)) => println!("任务2完成: {}", result),
        Ok(Err(e)) => eprintln!("任务2执行出错: {}", e),
        Err(_) => {
            eprintln!("任务2超时，已终止");
            task2_handle.abort();
        }
    }

    Ok(())
}

方案2：CancellationToken 多任务联动超时取消 #

适合需要统一控制多任务取消的场景（比如超时后所有任务一起终止）：

首先在Cargo.toml添加依赖：

tokio = { version = "1.0", features = ["full"] }
tokio-util = { version = "0.7", features = ["sync"] }
reqwest = "0.11"

代码示例：

use tokio::time::{sleep, Duration};
use tokio_util::sync::CancellationToken;
use tokio::spawn;
use std::error::Error;

async fn call_rest_api(api_url: &str, cancel_token: CancellationToken) -> Result<String, Box<dyn Error>> {
    // 通过select!同时监听API响应和取消信号
    tokio::select! {
        resp = reqwest::get(api_url) => {
            let text = resp?.text().await?;
            Ok(text)
        }
        _ = cancel_token.cancelled() => {
            Err("任务被取消".into())
        }
    }
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let cancel_token = CancellationToken::new();
    let timeout_token = cancel_token.clone();

    // 启动超时触发任务，5秒后取消所有关联任务
    spawn(async move {
        sleep(Duration::from_secs(5)).await;
        timeout_token.cancel();
        println!("超时时间到，触发所有任务取消");
    });

    // 启动两个API任务，传入同一取消令牌
    let task1 = spawn(call_rest_api("https://api.example.com/service1", cancel_token.clone()));
    let task2 = spawn(call_rest_api("https://api.example.com/service2", cancel_token.clone()));

    // 等待任务完成或被取消
    match task1.await {
        Ok(Ok(res)) => println!("任务1结果: {}", res),
        Ok(Err(e)) => eprintln!("任务1被取消/出错: {}", e),
        Err(_) => eprintln!("任务1执行异常"),
    }

    match task2.await {
        Ok(Ok(res)) => println!("任务2结果: {}", res),
        Ok(Err(e)) => eprintln!("任务2被取消/出错: {}", e),
        Err(_) => eprintln!("任务2执行异常"),
    }

    Ok(())
}

关键注意事项 #

• 任务内部必须包含可取消的await点：如果有长时间同步阻塞代码，需要放到tokio::task::spawn_blocking中，否则Tokio无法中断该任务。
• timeout返回Err时，任务已被Tokio标记为取消，但调用abort()能确保任务立即停止调度。
• 使用CancellationToken时，必须在任务内部通过select!监听取消信号，才能及时响应取消操作。

内容的提问来源于stack exchange，提问作者Mohamad-Jaafar NEHME