Rust异步工具库tokio-utils的使用指南

tokio-utils 是 Tokio 生态系统中的一个实用工具库，为异步编程提供了许多高效实用的辅助功能。这个库包含了许多在异步编程中常见问题的解决方案，可以帮助开发者编写更简洁、更高效的异步代码。

主要功能

tokio-utils 提供了以下几类实用工具：

1. 异步原语：扩展了 Tokio 的基础异步原语
2. Future 工具：提供了各种 Future 组合器和实用程序
3. Stream 工具：增强了 Stream 的处理能力
4. IO 工具：提供了各种 IO 相关的实用功能

安装

在 Cargo.toml 中添加依赖：

[dependencies]
tokio-utils = "0.7"
tokio = { version = "1.0", features = ["full"] }

核心功能示例

1. 定时器工具

use tokio_utils::time::{delay_for, Delay};

#[tokio::main]
async fn main() {
    // 延迟500毫秒
    delay_for(std::time::Duration::from_millis(500)).await;
    println!("500ms后执行");
    
    // 创建可取消的延迟
    let delay = Delay::new(std::time::Duration::from_secs(1));
    tokio::select! {
        _ = delay => println!("1秒后执行"),
        _ = tokio::signal::ctrl_c() => println!("取消延迟"),
    }
}

2. Future 工具

use tokio_utils::future::{FutureExt, OptionFuture};

#[tokio::main]
async fn main() {
    // 将Option转换为Future
    let some_future = OptionFuture::from(Some(async { 42 }));
    let none_future = OptionFuture::<i32>::from(None);
    
    assert_eq!(some_future.await, Some(42));
    assert_eq!(none_future.await, None);
    
    // Future的扩展方法
    let future = async { 1 };
    let result = future.map(|x| x + 1).await;
    assert_eq!(result, 2);
}

3. Stream 工具

use tokio_utils::stream::{StreamExt, MergeStreams};
use tokio::stream::{self, Stream};

#[tokio::main]
async fn main() {
    // 合并多个Stream
    let s1 = stream::iter(vec![1, 2, 3]);
    let s2 = stream::iter(vec![4, 5, 6]);
    
    let merged = MergeStreams::new(vec![s1, s2]);
    let results: Vec<_> = merged.collect().await;
    assert!(results.contains(&1) && results.contains(&6));
    
    // Stream扩展方法
    let s = stream::iter(1..=3);
    let doubled: Vec<_> = s.map(|x| x * 2). collect().await;
    assert_eq!(doubled, vec![2, 4, 6]);
}

4. 同步工具

use tokio_utils::sync::{CancellationToken, Semaphore};

#[tokio::main]
async fn main() {
    // 取消令牌
    let token = CancellationToken::new();
    let child_token = token.child_token();
    
    tokio::spawn(async move {
        tokio::select! {
            _ = child_token.cancelled() => println!("任务被取消"),
            _ = tokio::time::delay_for(std::time::Duration::from_secs(1)) => {
                println!("任务完成")
            }
        }
    });
    
    // 模拟取消
    token.cancel();
    
    // 信号量
    let semaphore = Semaphore::new(3);
    for i in 0..5 {
        let permit = semaphore.acquire().await.unwrap();
        tokio::spawn(async move {
            println!("任务{}开始", i);
            tokio::time::delay_for(std::time::Duration::from_secs(1)).await;
            drop(permit); // 释放许可
            println!("任务{}完成", i);
        });
    }
}

高级用法

自定义 Future 组合器

use tokio_utils::future::FutureExt;
use std::time::Duration;

#[tokio::main]
async fn main() {
    let future = async {
        tokio::time::sleep(Duration::from_secs(1)).await;
        42
    };
    
    // 添加超时
    match future.timeout(Duration::from_millis(500)).await {
        Ok(result) => println!("成功: {}", result),
        Err(_) => println!("超时"),
    }
}

批量处理 Stream

use tokio_utils::stream::StreamExt;
use tokio::stream;

#[tokio::main]
async fn main() {
    let stream = stream::iter(0..10);
    
    // 每3个元素一批处理
    let batches = stream.chunks(3);
    
    tokio::pin!(batches);
    while let Some(batch) = batches.next().await {
        println!("批处理: {:?}", batch);
    }
}

完整示例

下面是一个整合了tokio-utils多个功能的完整示例：

use tokio_utils::{
    future::FutureExt,
    stream::StreamExt,
    sync::{CancellationToken, Semaphore},
    time::{delay_for, Delay}
};
use tokio::stream::{self, Stream};
use std::time::Duration;

#[tokio::main]
async fn main() {
    // 1. 定时器示例
    println!("--- 定时器示例 ---");
    delay_for(Duration::from_millis(300)).await;
    println!("300ms延迟后执行");
    
    let delay = Delay::new(Duration::from_secs(1));
    tokio::select! {
        _ = delay => println!("1秒延迟完成"),
        _ = tokio::signal::ctrl_c() => println!("取消了延迟"),
    }

    // 2. Future工具示例
    println!("\n--- Future工具示例 ---");
    let future = async {
        delay_for(Duration::from_millis(800)).await;
        100
    };
    
    match future.timeout(Duration::from_millis(500)).await {
        Ok(val) => println!("在超时前完成: {}", val),
        Err(_) => println!("操作超时"),
    }

    // 3. Stream工具示例
    println!("\n--- Stream工具示例 ---");
    let s1 = stream::iter(vec!["a", "b", "c"]);
    let s2 = stream::iter(vec![1, 2, 3]);
    
    let merged = s1.zip(s2);
    let results: Vec<_> = merged.collect().await;
    println!("合并后的结果: {:?}", results);

    // 4. 同步工具示例
    println!("\n--- 同步工具示例 ---");
    let token = CancellationToken::new();
    let child = token.child_token();
    
    let task = tokio::spawn(async move {
        tokio::select! {
            _ = child.cancelled() => println!("子任务被取消"),
            _ = delay_for(Duration::from_secs(2)) => println!("子任务完成"),
        }
    });
    
    delay_for(Duration::from_millis(500)).await;
    token.cancel();
    task.await.unwrap();
    
    let sem = Semaphore::new(2);
    for i in 0..4 {
        let permit = sem.acquire().await.unwrap();
        tokio::spawn(async move {
            println!("任务 {} 开始", i);
            delay_for(Duration::from_secs(1)).await;
            drop(permit);
            println!("任务 {} 完成", i);
        });
    }
    
    delay_for(Duration::from_secs(2)).await;
}

最佳实践

1. 合理使用取消令牌：对于需要取消的长时运行任务，使用 CancellationToken 可以更优雅地处理取消逻辑

2. 控制并发：使用 Semaphore 限制资源密集型操作的并发数量

3. 组合Future：利用提供的组合器减少样板代码，使异步逻辑更清晰

4. 处理超时：为所有可能长时间运行的操作添加超时处理

tokio-utils 通过提供这些实用工具，大大简化了 Tokio 异步编程中的常见模式实现，让开发者能更专注于业务逻辑而非基础设施代码。