跳转至

11. 并发篇

对应章节：第16章（并发）。
本篇掌握 Rust 的并发工具：线程、消息传递、共享状态。

项目 18：多线程下载器

目标：多线程并发下载 URL，掌握 thread / channel / Arc / Mutex。

创建项目

cargo new downloader
cd downloader
cargo add reqwest --features blocking   # 添加最新版 reqwest，启用 blocking 特性

功能要求

cargo run -- 4 https://example.com/a.html https://example.com/b.html https://example.com/c.html
  • 第一个参数：并发数（线程数）
  • 后续参数：URL 列表，均匀分配到各线程
  • 线程用 channel 把结果发回主线程
  • 打印进度（Mutex 计数器）和总耗时

知识点

知识点 说明
thread::spawn 创建线程
move 闭包 所有权移入线程
JoinHandle .join() 等待线程结束
mpsc::channel 多生产者单消费者通道
Sender::clone 多发送端
Arc<T> 线程安全版 Rc（原子计数）
Mutex<T> 线程安全版 RefCell（加锁）
Instant 计时

代码

依赖已由 cargo add reqwest --features blocking 添加（cargo search reqwest 可查最新版本），无需手动编辑 Cargo.toml。

src/main.rs：

use std::env;
use std::sync::{mpsc, Arc, Mutex};
use std::thread;
use std::time::Instant;

// 错误统一转成 String 装进 Result——错误也是一种返回值
fn download(url: &str) -> Result<String, String> {
    let resp = reqwest::blocking::get(url).map_err(|e| e.to_string())?;
    let body = resp.bytes().map_err(|e| e.to_string())?;
    Ok(format!("已下载：{}（{} 字节）", url, body.len()))
}

fn main() {
    let args: Vec<String> = env::args().collect();
    let n_threads: usize = args[1].parse().unwrap();
    let urls: Vec<String> = args[2..].to_vec();

    let start = Instant::now();   // 计时起点

    // channel：tx 发送结果回主线程，rx 主线程接收
    // mpsc = Multiple Producer, Single Consumer（多生产单消费）
    let (tx, rx) = mpsc::channel();

    // Arc 让所有线程共享 URL 列表（不能直接传引用，线程可能活得比 main 久）
    let url_list = Arc::new(urls);
    let url_count = url_list.len();

    // Mutex 计数器：所有线程共享并安全修改
    let counter = Arc::new(Mutex::new(0));

    let mut handles = Vec::new();

    // 创建 n_threads 个线程
    for thread_id in 0..n_threads {
        // 每个线程都要克隆一份 tx / url_list / counter
        let tx = tx.clone();
        let url_list = Arc::clone(&url_list);
        let counter_clone = Arc::clone(&counter);

        // move：把捕获变量的所有权移入线程闭包
        let handle = thread::spawn(move || {
            // 按 thread_id 均匀分配：id, id+n, id+2n, ...
            // 保证每个 URL 恰好被一个线程下载
            let mut i = thread_id;
            while i < url_count {
                let url = &url_list[i];
                let result = download(url);
                tx.send(result).unwrap();   // 结果（含错误）发回主线程

                // 加锁修改共享计数器
                let mut count = counter_clone.lock().unwrap();   // lock 返回 MutexGuard
                *count += 1;                                     // 解引用后修改
                println!("进度: {}/{}", *count, url_count);
                i += n_threads;
            }
        });
        handles.push(handle);
    }

    drop(tx);   // 释放主线程的发送端，rx 才能迭代结束（否则死锁）

    // 接收所有结果（阻塞直到所有发送端关闭）
    for received in rx {
        println!("Received: {:?}", received);   // Result 的 Debug 形如 Ok("已下载…")
    }

    // 等待所有线程结束（确保全部任务完成）
    for handle in handles {
        handle.join().unwrap();
    }

    println!("Total of time: {:?}", start.elapsed());
}

运行

cargo run -- 3 https://example.com https://example.org https://example.net
# 进度: 1/3
# Received: Ok("已下载：https://example.com（1256 字节）")
# ...
# Total of time: 301.5ms   ← 三个请求并发，耗时接近单个而非三倍

注意事项（重点）

  1. move 闭包：thread::spawn 的闭包可能活得比主线程变量久，必须 move 把捕获变量所有权移入线程；若只借用引用会报 borrowed value does not live long enough。Vec<&String> 这种借用进线程必错，要克隆成 Vec<String> 让线程拥有数据。
  2. Arc vs Rc：Rc 计数非原子，跨线程不安全；Arc 用原子操作，可安全共享。
  3. Mutex vs RefCell：Mutex 是线程安全版内部可变性；lock() 返回 guard，*count += 1 需解引用。忘记 lock 共享可变会编译失败（避免数据竞争）。
  4. drop(tx) 必须：主线程保留一个 tx，for x in rx 永远等不到发送端全部关闭，会死锁。显式 drop(tx) 让迭代正常结束。
  5. 均匀分配：i = thread_id; i += n_threads 让每个 URL 恰好被一个线程下载，避免"每线程下载全部 URL"的重复。
  6. Arc::clone 是浅克隆：只加计数，不复制 URL 数据。
  7. Result 作为消息：download 把两类错误都 map_err(|e| e.to_string()) 转成 String，统一装进 Result 发回主线程——错误也是数据，接收端拿到 Ok(...)/Err(...) 统一打印即可，单个下载失败不会 panic 中断其他线程。
  8. 改进练习：当前不带参数运行会因 args[1] 越界 panic——自己加回参数检查与用法提示。

评论