Tokio异步请求间数据共享及死链检查代码优化咨询

Great work putting together an async link checker with Tokio and reqwest! Let's break down how to refine this code to be more robust, idiomatic, and maintainable—while making sure you reliably capture every request's outcome.

Key Observations & Improvements #

1. Ditch Legacy Tokio APIs for Modern Async Patterns #

Your current code uses older Tokio constructs like stream::iter_ok and tokio::run, which have been replaced by more ergonomic APIs in recent versions. Let's update to the latest patterns:

• Use #[tokio::main] macro instead of tokio::run to simplify setting up the async runtime.
• Use tokio_stream::StreamExt to work with async streams (replace stream::iter_ok with tokio_stream::iter).

2. Fix Error Handling (No More unwrap()!) #

Calling unwrap() on channel sends or client operations will crash your program if something goes wrong (e.g., the receiver drops before all sends complete). Instead, handle errors gracefully:

• Check channel send results and log failures instead of panicking.
• Capture and propagate detailed error information for failed requests (not just format it).

3. Capture Meaningful Results, Not Just Success Flags #

Right now you're sending 1 for successful requests, but you'll want to track which URLs succeeded/failed and why. Define a clear result type to encapsulate this information.

4. Clean Up Ownership & Resource Management #

• Ensure the reqwest Client is properly shared (you're already doing this right—good job! Reusing a single client is critical for connection pooling).
• Avoid unnecessary clones of the channel sender where possible, and use channel sizing to manage backpressure.

Refactored Code Example #

Here's how your code might look with these improvements:

use reqwest::Client;
use tokio::sync::mpsc;
use tokio_stream::{StreamExt, wrappers::ReceiverStream};

// Define a type to capture detailed request results
#[derive(Debug)]
enum CheckResult {
    Success(String, u16), // URL + status code
    Failure(String, String), // URL + error message
}

const PARALLEL_REQUESTS: usize = 10;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let urls = vec![
        "https://example.com",
        "https://invalid-url.example",
        "https://github.com",
    ];

    // Create a channel with a buffer size matching the number of requests
    let (tx, rx) = mpsc::channel(urls.len());

    // Initialize async reqwest client
    let client = Client::new();

    // Process URLs as an async stream
    let task = tokio::spawn(async move {
        tokio_stream::iter(urls)
            .map(|url| {
                let client = client.clone();
                let tx = tx.clone();
                async move {
                    match client.get(url).send().await {
                        Ok(response) => {
                            let status = response.status().as_u16();
                            // Send success result
                            if let Err(e) = tx.send(CheckResult::Success(url.to_string(), status)).await {
                                eprintln!("Failed to send success result for {}: {}", url, e);
                            }
                        }
                        Err(e) => {
                            // Send failure result with error details
                            if let Err(send_err) = tx.send(CheckResult::Failure(url.to_string(), e.to_string())).await {
                                eprintln!("Failed to send failure result for {}: {}", url, send_err);
                            }
                        }
                    }
                }
            })
            .buffer_unordered(PARALLEL_REQUESTS)
            .for_each(|_| async {})
            .await;
    });

    // Wait for all requests to complete
    task.await?;

    // Collect and process results from the channel
    let results = ReceiverStream::new(rx).collect::<Vec<_>>().await;

    // Print summary
    let (success_count, failure_count) = results.iter().fold((0, 0), |(s, f), res| {
        match res {
            CheckResult::Success(url, status) => {
                println!("✅ {} - Status: {}", url, status);
                (s + 1, f)
            }
            CheckResult::Failure(url, err) => {
                println!("❌ {} - Error: {}", url, err);
                (s, f + 1)
            }
        }
    });

    println!("\nSummary: {} successful, {} failed", success_count, failure_count);

    Ok(())
}

Additional Notes #

• Connection Pooling: Reusing a single reqwest::Client is essential—it maintains a pool of HTTP connections, which drastically improves performance for multiple requests.
• Backpressure: Using buffer_unordered(PARALLEL_REQUESTS) limits the number of concurrent requests, preventing overwhelming the target servers or your own network.
• Result Collection: By using ReceiverStream to collect results, you ensure you capture every outcome before exiting, even if some requests take longer than others.

内容的提问来源于stack exchange，提问作者Artir