rust-tokio-semaphore-concurrency

为什么需要信号量

tokio 下用 join! 或 spawn 一堆任务很轻松，但一次性并发几百个网络请求会把下游打爆，本地文件句柄和内存也扛不住。信号量就是一把限流闸：预先放 N 个许可，谁拿到许可谁才能开工，干完归还，并发数就被钉在 N 以内。

基础用法

Semaphore::new(3) 表示最多 3 个任务同时执行。用 acquire_owned 拿到一个可移动的许可，方便带进 async 任务。

use tokio::sync::Semaphore;
use std::sync::Arc;

async fn do_work(i: u32) {
    // 模拟一次耗时请求
    tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    println!("done {i}");
}

#[tokio::main]
async fn main() {
    let sem = Arc::new(Semaphore::new(3));

    let mut handles = vec![];
    for i in 0..10 {
        let permit = sem.clone().acquire_owned().await.unwrap();
        handles.push(tokio::spawn(async move {
            do_work(i).await;
            drop(permit); // 显式归还，离开作用域也会自动归还
        }));
    }

    for h in handles {
        let _ = h.await;
    }
}

在并发抓取里限流

把信号量带进抓取循环，每个 URL 先拿许可再 spawn，即使有上千个链接，同时在飞的网络请求也始终不超过 limit。

use tokio::sync::Semaphore;
use std::sync::Arc;

async fn fetch(url: String) -> Result<String, reqwest::Error> {
    Ok(reqwest::get(&url).await?.text().await?)
}

async fn fetch_all(urls: Vec<String>, limit: usize) -> Vec<Result<String, reqwest::Error>> {
    let sem = Arc::new(Semaphore::new(limit));
    let mut tasks = Vec::with_capacity(urls.len());

    for url in urls {
        let permit = sem.clone().acquire_owned().await.unwrap();
        tasks.push(tokio::spawn(async move {
            let result = fetch(url).await;
            drop(permit);
            result
        }));
    }

    let mut out = Vec::with_capacity(tasks.len());
    for t in tasks {
        out.push(t.await.unwrap());
    }
    out
}

注意点

• 普通 acquire 借用的许可不能跨 await 带进 spawn 的任务，必须用 acquire_owned 拿一个可拥有的许可。
• 许可靠 drop 或离开作用域释放，忘记归还就会永久少一个配额，并发上限慢慢被吃掉。
• limit 大小看下游承受能力，爬虫一般 5 到 20，压测接口可以更小。