Rust 异步测试与调试实战指南

异步操作中错误处理同样关键。我们需要验证成功路径和异常分支。

// src/lib.rs
use std::io;
use tokio::time::sleep;
use std::time::Duration;

pub async fn read_file(path: &str) -> Result<String, io::Error> {
    sleep(Duration::from_millis(100)).await;
    if path == "invalid" {
        return Err(io::Error::new(io::ErrorKind::NotFound, "File not found"));
    }
    Ok("Hello, World!".to_string())
}

测试用例需覆盖正常返回和错误类型：

// tests/lib.rs
use my_crate::read_file;
use tokio::test;
use std::io;

#[test]
async fn test_read_file_success() {
    let result = read_file("valid.txt").await;
    assert!(result.is_ok());
    assert_eq!(result.unwrap(), "Hello, World!");
}

#[test]
async fn test_read_file_error() {
    let result = read_file("invalid").await;
    assert!(result.is_err());
    assert_eq!(result.unwrap_err().kind(), io::ErrorKind::NotFound);
}

1.5 异步超时测试

对于耗时操作，防止测试挂起是必须的。我们可以利用 tokio::time::timeout 包装任务。

// src/lib.rs
use tokio::time::sleep;
use std::time::Duration;

pub async fn long_running_task() {
    sleep(Duration::from_secs(5)).await;
}

测试中设置更短的超时时间：

// tests/lib.rs
use my_crate::long_running_task;
use tokio::test;
use tokio::time::timeout;
use std::time::Duration;

#[test]
async fn test_long_running_task_timeout() {
    let result = timeout(Duration::from_secs(3), long_running_task()).await;
    assert!(result.is_err());
}

二、异步集成测试

2.1 服务间通信测试

集成测试通常涉及真实的服务调用。这里演示如何测试 HTTP 健康检查接口。

// tests/integration.rs
use tokio::test;
use reqwest::Client;

#[test]
async fn test_user_sync_service() {
    let client = Client::new();
    let response = client.get("http://localhost:3000/health")
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), 200);
}

#[test]
async fn test_order_processing_service() {
    let client = Client::new();
    let response = client.get("http://localhost:3001/health")
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), 200);
}

2.2 数据库操作测试

数据库连接池的初始化也是集成测试的重点。确保连接可用且表结构正确。

// tests/integration.rs
use tokio::test;
use sqlx::PgPool;
use my_crate::db;

#[test]
async fn test_create_pool() {
    let config = db::DbConfig {
        url: "postgresql://test:test@localhost:5432/test_db".to_string(),
    };
    let pool = db::create_pool(config).await.unwrap();
    assert!(pool.is_connected().await);
}

#[test]
async fn test_create_table() {
    let config = db::DbConfig {
        url: "postgresql://test:test@localhost:5432/test_db".to_string(),
    };
    let pool = db::create_pool(config).await.unwrap();
    let result = sqlx::query!("CREATE TABLE IF NOT EXISTS users (id SERIAL PRIMARY KEY, name TEXT, email TEXT)")
        .execute(&pool)
        .await;
    assert!(result.is_ok());
}

2.3 Redis 操作测试

Redis 测试通常关注连接建立和基本命令执行。

// tests/integration.rs
use tokio::test;
use redis::Client;
use my_crate::redis;

#[test]
async fn test_set_get() {
    let config = redis::RedisConfig {
        url: "redis://localhost:6379/0".to_string(),
    };
    let client = redis::create_client(config).await.unwrap();
    let mut conn = client.get_connection().await.unwrap();
    
    let key = "test_key";
    let value = "test_value";
    redis::cmd("SET").arg(key).arg(value)
        .query_async(&mut conn).await.unwrap();
    
    let result: String = redis::cmd("GET").arg(key)
        .query_async(&mut conn).await.unwrap();
    assert_eq!(result, value);
}

2.4 外部 API 依赖模拟

直接调用第三方 API 不稳定，推荐使用 wiremock 进行本地模拟。

// src/http.rs
use reqwest::Client;
use serde::Deserialize;

#[derive(Deserialize, Debug)]
pub struct User {
    pub id: i32,
    pub name: String,
    pub email: String,
}

pub struct UserApiClient {
    client: Client,
    base_url: String,
}

impl UserApiClient {
    pub fn new(base_url: &str) -> Self {
        UserApiClient {
            client: Client::new(),
            base_url: base_url.to_string(),
        }
    }

    pub async fn get_user(&self, id: i32) -> Result<User, reqwest::Error> {
        self.client.get(&format!("{}/users/{}", self.base_url, id))
            .send().await?.json().await
    }
}

测试中使用 MockServer 拦截请求：

// tests/integration.rs
use tokio::test;
use wiremock::matchers::{method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};
use my_crate::http::UserApiClient;

#[test]
async fn test_get_user() {
    let mock_server = MockServer::start().await;
    let client = UserApiClient::new(&mock_server.uri());
    let user_id = 1;
    
    let mock_response = serde_json::json!({
        "id": user_id,
        "name": "Test User",
        "email": "[email protected]"
    });
    
    Mock::given(method("GET"))
        .and(path(format!("/users/{}", user_id)))
        .respond_with(ResponseTemplate::new(200).set_body_json(mock_response))
        .mount(&mock_server)
        .await;
    
    let user = client.get_user(user_id).await.unwrap();
    assert_eq!(user.id, user_id);
    assert_eq!(user.name, "Test User");
}

三、异步性能测试

3.1 性能测试概念

性能测试关注响应时间、吞吐量和资源利用率。异步系统需特别留意并发度对任务调度的影响。

3.2 常用工具

• Wrk：高并发 HTTP 压测工具。
• K6：脚本化 API 性能测试，支持实时监控。
• Apache Benchmark (ab)：轻量级 HTTP 基准测试。
• Locust：分布式 Python 压测框架。

3.3 API 接口压测

使用 Wrk 进行快速压测：

# 测试 GET 接口，并发 100 个用户，持续 10 秒
wrk -t12 -c100 -d10s "http://localhost:3000/api/users"

使用 K6 编写脚本：

// k6 脚本
import http from 'k6/http';
import { check, group } from 'k6';

export let options = {
    vus: 100, // 并发用户数
    duration: '10s', // 测试持续时间
};

export default function () {
    group('Test Users API', function () {
        let response = http.get('http://localhost:3000/api/users');
        check(response, {
            'status is 200': (r) => r.status === 200,
            'response time < 500ms': (r) => r.timings.duration < 500,
        });
    });
}

3.4 数据库与 Redis 压测

数据库操作同样需要独立评估。Redis 可使用原生工具：

# 测试 Redis 的 GET 操作
redis-benchmark -h localhost -p 6379 -t get -n 100000 -c 100

四、异步调试工具的使用

4.1 日志系统配置

合理的日志是排查问题的第一步。结合 log 和 env_logger 即可快速搭建。

// src/lib.rs
use log::{info, error};

pub async fn foo() {
    info!("Entering foo function");
    let result = bar().await;
    if result.is_err() {
        error!("Bar function failed: {:?}", result);
    }
}

async fn bar() -> Result<(), String> {
    Ok(())
}

主程序中初始化日志级别：

// src/main.rs
use my_crate::foo;
use log::LevelFilter;
use simple_logger::SimpleLogger;

#[tokio::main]
async fn main() {
    SimpleLogger::new().with_level(LevelFilter::Info).init().unwrap();
    foo().await;
}

4.2 Tokio Console 使用

Tokio Console 能可视化监控任务执行、I/O 和资源占用。使用前需开启未稳定特性。

// src/main.rs
use log::{info, error};
use tokio::time::sleep;
use std::time::Duration;
use tracing_subscriber::prelude::*;
use tracing::info as trace_info;

#[tokio::main]
async fn main() {
    tracing_subscriber::registry()
        .with(tracing_subscriber::EnvFilter::new("info"))
        .with(tracing_subscriber::fmt::layer())
        .init();
    
    trace_info!("Application started");
    let mut handles = Vec::new();
    for i in 1..=3 {
        let handle = tokio::spawn(async move {
            trace_info!("Task {} started", i);
            sleep(Duration::from_millis(100 * i)).await;
            trace_info!("Task {} finished", i);
            i
        });
        handles.push(handle);
    }
    let results: Vec<_> = futures::future::join_all(handles)
        .await.into_iter()
        .map(|r| r.unwrap())
        .collect();
    trace_info!("All tasks finished. Results: {:?}", results);
}

启动命令：

RUSTFLAGS="--cfg tokio_unstable" RUST_LOG=info cargo run --bin tokio-console

4.3 GDB 和 LLDB 调试

底层调试依然有效，但需注意异步上下文切换。

// src/main.rs
use tokio::time::sleep;
use std::time::Duration;

#[tokio::main]
async fn main() {
    let handle = tokio::spawn(async move {
        println!("Task started");
        sleep(Duration::from_millis(100)).await;
        println!("Task finished");
    });
    handle.await.unwrap();
}

调试步骤：

rust-gdb target/debug/my_crate
(gdb) break main
(gdb) run

4.4 内存泄漏检测

Valgrind 和 AddressSanitizer 是检测内存问题的利器。

// src/main.rs
use tokio::time::sleep;
use std::time::Duration;
use std::sync::Arc;

struct MyData {
    value: i32,
}

impl Drop for MyData {
    fn drop(&mut self) {
        println!("MyData dropped");
    }
}

#[tokio::main]
async fn main() {
    let data = Arc::new(MyData { value: 42 });
    let handle = tokio::spawn(async move {
        println!("Task running");
        sleep(Duration::from_millis(100)).await;
        println!("Task finished");
    });
    handle.await.unwrap();
    println!("Main task finished");
}

检测命令：

# Valgrind
cargo run --release -- --test valgrind --leak-check=yes

# AddressSanitizer
RUSTFLAGS="-Z sanitizer=address" cargo run --release

五、实战项目的测试与调试

5.1 用户同步服务测试

// tests/integration.rs
use tokio::test;
use reqwest::Client;
use serde_json::json;

#[test]
async fn test_sync_users() {
    let client = Client::new();
    let response = client.post("http://localhost:3000/api/users/sync")
        .json(&json!({"limit": 1000}))
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), 202);
    
    let response = client.get("http://localhost:3000/health")
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), 200);
}

5.2 订单处理服务测试

// tests/integration.rs
use tokio::test;
use reqwest::Client;
use serde_json::json;

#[test]
async fn test_process_order() {
    let client = Client::new();
    let response = client.post("http://localhost:3001/api/orders/process")
        .json(&json!({"user_id": 1, "product_id": 2, "quantity": 3}))
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), 200);
    
    let response = client.get("http://localhost:3001/health")
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), 200);
}

5.3 监控服务 WebSocket 测试

// tests/integration.rs
use tokio::test;
use reqwest::Client;

#[test]
async fn test_websocket_connection() {
    let client = Client::new();
    let response = client.get("http://localhost:3002/health")
        .send()
        .await
        .unwrap();
    assert_eq!(response.status(), 200);
    
    let (socket, _) = tokio_tungstenite::connect_async("ws://localhost:3002/ws")
        .await
        .unwrap();
    socket.send(tokio_tungstenite::tungstenite::Message::Text("ping".to_string()))
        .await
        .unwrap();
    
    let message = socket.next().await.unwrap().unwrap();
    assert_eq!(message.to_text().unwrap(), "pong");
}

5.4 调试案例：任务失败定位

假设同步服务出现任务崩溃，可通过 Tokio Console 观察任务状态变化。

// src/main.rs
use log::{info, error};
use tokio::time::sleep;
use std::time::Duration;
use tracing_subscriber::prelude::*;
use tracing::info as trace_info;

#[tokio::main]
async fn main() {
    tracing_subscriber::registry()
        .with(tracing_subscriber::EnvFilter::new("info"))
        .with(tracing_subscriber::fmt::layer())
        .init();
    
    trace_info!("Application started");
    let mut handles = Vec::new();
    for i in 1..=3 {
        let handle = tokio::spawn(async move {
            trace_info!("Task {} started", i);
            if i == 2 {
                panic!("Task {} failed", i); // 模拟任务失败
            }
            sleep(Duration::from_millis(100 * i)).await;
            trace_info!("Task {} finished", i);
            i
        });
        handles.push(handle);
    }
    let results: Vec<_> = futures::future::join_all(handles)
        .await.into_iter()
        .map(|r| r.unwrap())
        .collect();
    trace_info!("All tasks finished. Results: {:?}", results);
}

配合环境变量启动调试：

RUSTFLAGS="--cfg tokio_unstable" RUST_LOG=info cargo run --bin tokio-console

六、测试与调试的最佳实践

6.1 测试驱动开发

TDD（测试驱动开发）在异步场景中尤为重要。先写测试用例能提前暴露异步竞态条件和错误处理遗漏。

6.2 代码覆盖率统计

使用 cargo-tarpaulin 量化测试覆盖情况：

cargo install cargo-tarpaulin
cargo tarpaulin --out Html

6.3 调试技巧总结

• 日志分级：生产环境关闭 Debug，保留 Info/Error。
• 工具辅助：善用 Tokio Console 和 ASan。
• 复现问题：确保能稳定触发 Bug。
• 隔离分析：将问题缩小到最小代码单元。

6.4 性能优化建议

• 任务调度：调整工作线程数匹配 CPU 核数。
• I/O 优化：使用连接池减少握手开销。
• 内存管理：避免频繁分配，考虑对象池复用。

七、总结

异步测试与调试是 Rust 开发中的核心挑战。掌握 tokio::test、集成测试框架、性能工具及调试手段，能显著提升系统的稳定性。希望本文提供的实战经验能帮助你在异步开发道路上走得更稳。