🦀 Rust 并发编程

// Send/Sync 是 auto trait —— 编译器自动推导
// 规则很简单：所有字段都 Send → 结构体 Send；所有字段都 Sync → 结构体 Sync

struct Safe { data: Vec<String> }  // ✅ 自动 Send + Sync

struct NotSync { cell: Cell<usize> }  // ❌ 不实现 Sync（Cell 不是 Sync）

struct NotSend { rc: Rc<String> }     // ❌ 不实现 Send（Rc 不是 Send）

// 手动标记（仅在确定安全时使用！）
struct ForceSend(*const u8);
unsafe impl Send for ForceSend {}  // unsafe：你向编译器保证它是安全的

use std::thread;
use std::time::Duration;

fn main() {
    // spawn 返回 JoinHandle<T>
    let handle = thread::spawn(|| {
        for i in 1..=5 {
            println!("子线程: 第 {} 次打印", i);
            thread::sleep(Duration::from_millis(100));
        }
        "子线程完成"  // 返回值
    });

    // 主线程继续执行
    for i in 1..=3 {
        println!("主线程: 第 {} 次打印", i);
        thread::sleep(Duration::from_millis(100));
    }

    // join 等待子线程结束，获取返回值
    let result = handle.join().unwrap();
    println!("子线程返回: {}", result);
}

use std::thread;

fn main() {
    let name = String::from("Alice");
    let age = 30;

    // ❌ 编译错误：name 的引用可能在主线程释放后仍在子线程中使用
    // thread::spawn(|| println!("Hello, {}", name));

    // ✅ move 把 name 和 age 的所有权转移到子线程
    let handle = thread::spawn(move || {
        println!("Hello, {}! You are {} years old.", name, age);
        // name 和 age 在这里被消费，主线程不能再使用
    });

    handle.join().unwrap();
}

use std::thread;

fn main() {
    let data = vec![1, 2, 3, 4, 5, 6, 7, 8];

    // scope 内创建的线程，在 scope 结束时自动 join
    thread::scope(|s| {
        // 切片引用可以安全传入，因为 scope 保证线程在 scope 结束前退出
        let chunk1 = &data[0..4];
        let chunk2 = &data[4..8];

        s.spawn(|| {
            let sum: i32 = chunk1.iter().sum();
            println!("前半求和: {}", sum);
        });

        s.spawn(|| {
            let sum: i32 = chunk2.iter().sum();
            println!("后半求和: {}", sum);
        });
    }); // ← 这里自动 join 所有线程，data 不会悬空

    println!("全部完成，data 仍可用: {:?}", data);
}

use rayon::prelude::*;  // 引入并行迭代器 trait

fn main() {
    let data: Vec<u64> = (1..=10_000_000).collect();

    // 单线程迭代
    let sum_seq: u64 = data.iter().sum();

    // 并行迭代 —— 只需把 iter() 改成 par_iter()
    let sum_par: u64 = data.par_iter().sum();

    assert_eq!(sum_seq, sum_par);

    // 更实用的例子：并行处理
    let results: Vec<String> = data
        .par_iter()
        .filter(|&&x| x % 2 == 0)
        .map(|&x| format!("even: {}", x))
        .collect();
}

use std::sync::mpsc;
use std::thread;
use std::time::Duration;

fn main() {
    // channel() 返回 (Sender, Receiver)
    let (tx, rx) = mpsc::channel();

    // 多生产者：clone sender
    let tx1 = tx.clone();
    let tx2 = tx.clone();

    // 生产者 1
    thread::spawn(move || {
        let msgs = vec!["任务A-1", "任务A-2", "任务A-3"];
        for msg in msgs {
            tx1.send(msg).unwrap();
            thread::sleep(Duration::from_millis(100));
        }
    });

    // 生产者 2
    thread::spawn(move || {
        let msgs = vec!["任务B-1", "任务B-2"];
        for msg in msgs {
            tx2.send(msg).unwrap();
            thread::sleep(Duration::from_millis(150));
        }
    });

    // drop 原始 sender，否则 rx.iter() 会永远阻塞
    drop(tx);

    // 消费者：迭代接收直到所有 sender 被 drop
    for received in rx.iter() {
        println!("收到: {}", received);
    }
    println!("所有生产者已断开，接收结束");
}

use crossbeam_channel as channel;
use std::thread;

fn main() {
    // unbounded
    let (tx, rx) = channel::unbounded();

    // bounded（支持 select!）
    let (tx_b, rx_b) = channel::bounded(100);

    // crossbeam 的 Receiver 实现了 Iterator
    // 更重要的是：Sender 和 Receiver 都是 Send + Sync
    // 可以在多线程间自由共享

    thread::spawn(move || {
        tx.send("hello from crossbeam").unwrap();
    });

    // recv() 阻塞接收，try_recv() 非阻塞
    let msg = rx.recv().unwrap();
    println!("{}", msg);

    // select! —— 同时等待多个 channel
    let (tx1, rx1) = channel::bounded(1);
    let (tx2, rx2) = channel::bounded(1);

    crossbeam_channel::select! {
        recv(rx1) -> msg => println!("从 rx1 收到: {:?}", msg),
        recv(rx2) -> msg => println!("从 rx2 收到: {:?}", msg),
    }
}

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    // Mutex 包裹需要保护的数据
    let counter = Arc::new(Mutex::new(0));

    let mut handles = vec![];
    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        let handle = thread::spawn(move || {
            // lock() 返回 MutexGuard，自动 unlock
            let mut num = counter.lock().unwrap();
            *num += 1;
            // MutexGuard 在这里 drop，自动释放锁
        });
        handles.push(handle);
    }

    for handle in handles {
        handle.join().unwrap();
    }

    println!("结果: {}", *counter.lock().unwrap());  // 10
}

use std::sync::{Arc, RwLock};
use std::thread;

fn main() {
    let data = Arc::new(RwLock::new(vec![1, 2, 3]));

    // 多个读者可以同时持有 read lock
    let mut readers = vec![];
    for i in 0..5 {
        let data = Arc::clone(&data);
        readers.push(thread::spawn(move || {
            let r = data.read().unwrap();  // RwLockReadGuard
            println!("读者 {}: {:?}", i, *r);
        }));
    }

    // 写者需要独占访问
    let data_w = Arc::clone(&data);
    let writer = thread::spawn(move || {
        let mut w = data.write().unwrap();  // RwLockWriteGuard
        w.push(4);
        println!("写者: 添加了元素");
    });

    for r in readers { r.join().unwrap(); }
    writer.join().unwrap();

    println!("最终数据: {:?}", *data.read().unwrap());
}

use std::sync::{Arc, Mutex};
use std::thread;

fn main() {
    // Arc: 让多个线程"共同拥有"同一个 Mutex
    // Mutex: 保护内部数据的安全访问
    let shared = Arc::new(Mutex::new(Vec::new()));

    let mut handles = vec![];
    for i in 0..4 {
        let shared = Arc::clone(&shared);  // 引用计数 +1（原子操作）
        handles.push(thread::spawn(move || {
            let mut data = shared.lock().unwrap();
            data.push(format!("来自线程 {}", i));
        }));
    }

    for h in handles { h.join().unwrap(); }
    println!("{:?}", *shared.lock().unwrap());
}

use std::sync::atomic::{AtomicUsize, AtomicBool, Ordering};
use std::sync::Arc;
use std::thread;

fn main() {
    let counter = Arc::new(AtomicUsize::new(0));
    let done = Arc::new(AtomicBool::new(false));

    let mut handles = vec![];
    for _ in 0..10 {
        let counter = Arc::clone(&counter);
        let done = Arc::clone(&done);
        handles.push(thread::spawn(move || {
            // fetch_add 返回旧值，原子地执行 += 1
            counter.fetch_add(1, Ordering::SeqCst);
        }));
    }

    for h in handles { h.join().unwrap(); }

    // load 原子地读取值
    println!("计数器: {}", counter.load(Ordering::SeqCst));  // 10

    // compare_exchange —— CAS 操作
    let old = 0;
    let new = 42;
    counter.compare_exchange(old, new, Ordering::SeqCst, Ordering::SeqCst);
}

use std::sync::{Arc, Barrier};
use std::thread;

fn main() {
    let n_threads = 4;
    let barrier = Arc::new(Barrier::new(n_threads));

    let mut handles = vec![];
    for i in 0..n_threads {
        let barrier = Arc::clone(&barrier);
        handles.push(thread::spawn(move || {
            println!("线程 {} 阶段1 完成", i);
            // 所有线程都到这里后，一起继续
            barrier.wait();
            println!("线程 {} 阶段2 开始", i);
        }));
    }

    for h in handles { h.join().unwrap(); }
}

use std::sync::{Arc, Mutex, Condvar};
use std::thread;
use std::time::Duration;

fn main() {
    // Condvar 总是和 Mutex 配对使用
    let pair = Arc::new((Mutex::new(false), Condvar::new()));
    let pair2 = Arc::clone(&pair);

    // 等待线程
    let waiter = thread::spawn(move || {
        let (lock, cvar) = &*pair2;
        let mut started = lock.lock().unwrap();
        // wait 会：1. 释放锁 2. 挂起 3. 被 notify 后重新获取锁
        while !*started {
            started = cvar.wait(started).unwrap();
        }
        println!("等待线程: 条件满足，继续执行");
    });

    // 通知线程
    thread::spawn(move || {
        let (lock, cvar) = &*pair;
        thread::sleep(Duration::from_secs(1));
        {
            let mut started = lock.lock().unwrap();
            *started = true;
        }
        // notify_one 唤醒一个等待线程
        cvar.notify_one();
        // cvar.notify_all() 唤醒所有等待线程
    }).join().unwrap();

    waiter.join().unwrap();
}

use std::sync::{OnceLock, Once};
use std::thread;

// OnceLock: 延迟初始化的全局变量
static DATABASE: OnceLock<String> = OnceLock::new();

fn get_database() -> &'static str {
    DATABASE.get_or_init(|| {
        println!("初始化数据库连接...");
        String::from("postgresql://localhost:5432/mydb")
    })
}

// Once: 只执行一次的代码块
static INIT: Once = Once::new();

fn initialize() {
    INIT.call_once(|| {
        println!("这段代码全局只执行一次");
    });
}

fn main() {
    // 多线程同时调用 get_database，只有第一个会真正初始化
    let mut handles = vec![];
    for _ in 0..5 {
        handles.push(thread::spawn(|| {
            let db = get_database();
            println!("获取到: {}", db);
        }));
    }
    for h in handles { h.join().unwrap(); }
}

pub trait Future {
    type Output;
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}

pub enum Poll<T> {
    Ready(T),     // 完成，返回结果
    Pending,      // 未完成，稍后再 poll
}

// async fn 返回一个 impl Future
async fn fetch_user(id: u32) -> User {
    let response = http_get(format!("/api/users/{}", id)).await;  // 等待 I/O
    let user: User = response.json().await;  // 等待解析
    user  // 返回值
}

// 等价于：
fn fetch_user(id: u32) -> impl Future<Output = User> {
    async move {
        let response = http_get(format!("/api/users/{}", id)).await;
        let user: User = response.json().await;
        user
    }
}

// async block —— 不需要函数，直接写
let future = async {
    let a = fetch_user(1).await;
    let b = fetch_user(2).await;
    (a, b)
};

// 当 Future 返回 Pending 时：
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Output> {
    if self.is_ready() {
        Poll::Ready(self.take_output())
    } else {
        // 把 waker 注册到 I/O 来源（epoll/kqueue/...）
        // 当 I/O 就绪时，epoll 会调用 waker.wake()
        // wake() 通知运行时重新 poll 这个 Future
        self.register_waker(cx.waker());
        Poll::Pending
    }
}

// 流程：
// 1. 运行时 poll Future → Pending
// 2. Future 注册 Waker
// 3. I/O 就绪 → Waker.wake() 被调用
// 4. 运行时重新 poll Future → Ready 或 Pending（循环）

#[tokio::main]  // 最常用：多线程运行时
async fn main() {
    println!("Hello, tokio!");
}

// 等价于手动创建：
fn main() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    rt.block_on(async {
        println!("Hello, tokio!");
    });
}

// 两种 Runtime：
// 1. 多线程（默认）—— 适合大多数场景，工作窃取调度
#[tokio::main]
async fn main() { /* ... */ }

// 2. 当前线程 —— 适合轻量/嵌入式场景
#[tokio::main(flavor = "current_thread")]
async fn main() { /* ... */ }

use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    // spawn 创建一个异步任务，返回 JoinHandle
    let handle = tokio::spawn(async {
        sleep(Duration::from_secs(1)).await;
        println!("异步任务完成");
        42
    });

    // 做其他事情...
    println!("主任务继续执行");

    // await JoinHandle 获取结果
    let result = handle.await.unwrap();  // 注意：.await 不是 .join()
    println!("结果: {}", result);
}

use tokio::sync::mpsc;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let (tx1, mut rx1) = mpsc::channel(32);
    let (tx2, mut rx2) = mpsc::channel(32);

    // 启动两个生产者
    tokio::spawn(async move {
        tx1.send("消息来自 channel 1").await.unwrap();
    });
    tokio::spawn(async move {
        sleep(Duration::from_millis(50)).await;
        tx2.send("消息来自 channel 2").await.unwrap();
    });

    // select! 谁先到处理谁
    tokio::select! {
        msg = rx1.recv() => {
            println!("收到 rx1: {:?}", msg);
        }
        msg = rx2.recv() => {
            println!("收到 rx2: {:?}", msg);
        }
    }

    // select! 循环 —— 处理多个事件流
    loop {
        tokio::select! {
            msg = rx1.recv() => {
                match msg {
                    Some(m) => println!("rx1: {}", m),
                    None => break,  // channel 关闭
                }
            }
            msg = rx2.recv() => {
                match msg {
                    Some(m) => println!("rx2: {}", m),
                    None => break,
                }
            }
        }
    }
}

use tokio::task::JoinSet;

#[tokio::main]
async fn main() {
    let mut set = JoinSet::new();

    // 批量 spawn
    for i in 0..5 {
        set.spawn(async move {
            tokio::time::sleep(Duration::from_millis(i * 100)).await;
            format!("任务 {} 完成", i)
        });
    }

    // 按完成顺序收集结果
    while let Some(result) = set.join_next().await {
        println!("{}", result.unwrap());
    }
}

use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    // 方式1: AbortHandle
    let handle = tokio::spawn(async {
        loop {
            sleep(Duration::from_secs(1)).await;
            println!("工作中...");
        }
    });

    let abort_handle = handle.abort_handle();
    // 5秒后取消
    sleep(Duration::from_secs(5)).await;
    abort_handle.abort();  // 或 handle.abort()
    let _ = handle.await;  // Err(JoinError::Cancelled)

    // 方式2: select! + 超时（更常用）
    let result = tokio::time::timeout(
        Duration::from_secs(3),
        some_async_operation()
    ).await;

    match result {
        Ok(value) => println!("完成: {:?}", value),
        Err(_) => println!("超时取消"),
    }
}

use tokio::sync::{mpsc, oneshot};

// Actor 的消息类型
enum CounterMsg {
    Increment,
    Decrement,
    Get(oneshot::Sender<i32>),  // 携带回复 channel
}

// Actor 结构体
struct CounterActor {
    receiver: mpsc::Receiver<CounterMsg>,
    count: i32,
}

impl CounterActor {
    fn new(receiver: mpsc::Receiver<CounterMsg>) -> Self {
        Self { receiver, count: 0 }
    }

    async fn run(&mut self) {
        while let Some(msg) = self.receiver.recv().await {
            match msg {
                CounterMsg::Increment => self.count += 1,
                CounterMsg::Decrement => self.count -= 1,
                CounterMsg::Get(reply) => {
                    let _ = reply.send(self.count);  // 回复当前值
                }
            }
        }
    }
}

// Actor 的 Handle（对外接口）
#[derive(Clone)]
struct CounterHandle {
    sender: mpsc::Sender<CounterMsg>,
}

impl CounterHandle {
    fn new() -> Self {
        let (sender, receiver) = mpsc::channel(128);
        let mut actor = CounterActor::new(receiver);
        tokio::spawn(async move { actor.run().await });
        Self { sender }
    }

    async fn increment(&self) {
        self.sender.send(CounterMsg::Increment).await.unwrap()
    }

    async fn get(&self) -> i32 {
        let (tx, rx) = oneshot::channel();
        self.sender.send(CounterMsg::Get(tx)).await.unwrap();
        rx.await.unwrap()
    }
}

#[tokio::main]
async fn main() {
    let counter = CounterHandle::new();
    counter.increment().await;
    counter.increment().await;
    println!("计数: {}", counter.get().await);  // 2
}

use tokio_stream::{self as stream, StreamExt};

async fn stream_example() {
    // 从迭代器创建 stream
    let mut s = stream::iter(1..=5);

    while let Some(value) = s.next().await {
        println!("收到: {}", value);
    }

    // 常用 stream 组合器
    let results: Vec<i32> = stream::iter(1..=100)
        .filter(|x| x % 2 == 0)       // 过滤偶数
        .map(|x| x * 3)                // 乘以3
        .take(5)                        // 取前5个
        .collect()
        .await;

    // tokio 提供的 stream
    // - LinesStream (按行读文件)
    // - BroadcastStream (广播 channel)
    // - WatchStream (watch channel)
    // - IntervalStream (定时器)
}

use tokio::sync::watch;

#[tokio::main]
async fn main() {
    let (tx, rx) = watch::channel("初始配置".to_string());

    // 多个消费者可以 clone Receiver
    let mut rx1 = rx.clone();
    let mut rx2 = rx.clone();

    // 消费者 1
    tokio::spawn(async move {
        while rx1.changed().await.is_ok() {
            let value = rx1.borrow();
            println!("消费者1: {}", *value);
        }
    });

    // 消费者 2
    tokio::spawn(async move {
        while rx2.changed().await.is_ok() {
            let value = rx2.borrow();
            println!("消费者2: {}", *value);
        }
    });

    // 生产者更新值
    tx.send("配置更新1".to_string()).unwrap();
    tx.send("配置更新2".to_string()).unwrap();
    tx.send("最终配置".to_string()).unwrap();
}

use std::sync::Arc;
use tokio::sync::Semaphore;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    const MAX_CONCURRENT: usize = 3;
    let semaphore = Arc::new(Semaphore::new(MAX_CONCURRENT));

    let mut handles = vec![];

    for i in 0..10 {
        let sem = Arc::clone(&semaphore);
        handles.push(tokio::spawn(async move {
            // 获取许可（如果已满则等待）
            let _permit = sem.acquire().await.unwrap();
            println!("任务 {} 开始执行（当前并发: {}）", i, MAX_CONCURRENT - sem.available_permits());
            sleep(Duration::from_secs(1)).await;
            println!("任务 {} 完成", i);
            // _permit drop 时自动释放许可
        }));
    }

    for h in handles { h.await.unwrap(); }
}

use std::sync::Mutex;

fn transfer(from: &Mutex<u64>, to: &Mutex<u64>, amount: u64) {
    let mut from_guard = from.lock().unwrap();
    // ⚠️ 竞态条件：在 check 和 use 之间，其他线程可能修改了 from
    if *from_guard >= amount {  // 检查
        *from_guard -= amount;  // 使用
        // 另一个线程在这里也可能通过了同样的检查
        drop(from_guard);  // 释放 from 锁
        *to.lock().unwrap() += amount;  // 获取 to 锁
    }
}
// 两个线程同时调用 transfer(A, B, 100) 和 transfer(A, C, 100)
// 如果 A 余额为 150，两个线程都通过了检查 → 超额转账！

use std::sync::Mutex;

// 方案1：手动保证顺序
fn transfer_safe(from: &Mutex<u64>, to: &Mutex<u64>, amount: u64,
                 from_id: usize, to_id: usize) {
    // 按 ID 顺序获取锁
    let (first, second) = if from_id < to_id {
        (from, to)
    } else {
        (to, from)
    };

    let mut first_guard = first.lock().unwrap();
    let mut second_guard = second.lock().unwrap();

    let (from_ref, to_ref) = if from_id < to_id {
        (&mut first_guard, &mut second_guard)
    } else {
        (&mut second_guard, &mut first_guard)
    };

    if **from_ref >= amount {
        **from_ref -= amount;
        **to_ref += amount;
    }
}

// 方案2：用一把大锁保护所有账户（简单但并发度低）
// 方案3：用 channel/message passing 替代锁

use parking_lot::Mutex;
use std::sync::Arc;

fn main() {
    // API 几乎一样，但 lock() 返回 MutexGuard（不需要 unwrap！）
    let data = Arc::new(Mutex::new(0));

    let data2 = Arc::clone(&data);
    std::thread::spawn(move || {
        let mut guard = data2.lock();  // 没有 .unwrap()
        *guard += 1;
        // guard drop 自动释放
    });

    // 带超时的尝试加锁
    if let Some(guard) = data.try_lock_for(std::time::Duration::from_secs(1)) {
        println!("获取锁成功: {}", *guard);
    } else {
        println!("1秒内未能获取锁");
    }
}

[dependencies]
tokio = { version = "1", features = ["full"] }
reqwest = { version = "0.12", features = ["rustls-tls"] }
tokio-stream = "0.1"

use std::collections::HashSet;
use std::sync::Arc;
use tokio::sync::{mpsc, Semaphore, Mutex, Notify};
use tokio::time::{timeout, Duration};

/// 爬取结果
#[derive(Debug)]
struct CrawlResult {
    url: String,
    status: u16,
    body_len: usize,
    links: Vec<String>,
}

/// 爬虫配置
struct CrawlerConfig {
    max_concurrent: usize,
    request_timeout: Duration,
    max_depth: u32,
    max_urls: usize,
}

impl Default for CrawlerConfig {
    fn default() -> Self {
        Self {
            max_concurrent: 10,
            request_timeout: Duration::from_secs(10),
            max_depth: 3,
            max_urls: 100,
        }
    }
}

/// 并发爬虫
struct Crawler {
    config: CrawlerConfig,
    visited: Arc<Mutex<HashSet<String>>>,
    semaphore: Arc<Semaphore>,
    client: reqwest::Client,
    shutdown: Arc<Notify>,
}

impl Crawler {
    fn new(config: CrawlerConfig) -> Self {
        let client = reqwest::Client::builder()
            .timeout(config.request_timeout)
            .user_agent("RustCrawler/1.0")
            .build()
            .unwrap();

        Self {
            semaphore: Arc::new(Semaphore::new(config.max_concurrent)),
            visited: Arc::new(Mutex::new(HashSet::new())),
            client,
            shutdown: Arc::new(Notify::new()),
            config,
        }
    }

    /// 启动爬虫，返回所有结果
    async fn crawl(&self, seed_urls: Vec<String>) -> Vec<CrawlResult> {
        let (url_tx, url_rx) = mpsc::channel<(String, u32)>(256);
        let (result_tx, mut result_rx) = mpsc::channel(256);

        // 发送种子 URL
        for url in seed_urls {
            let _ = url_tx.send((url, 0)).await;
        }

        // 工作线程：从 channel 取 URL，爬取，发送结果
        let worker_url_tx = url_tx.clone();
        let visited = Arc::clone(&self.visited);
        let semaphore = Arc::clone(&self.semaphore);
        let client = self.client.clone();
        let shutdown = Arc::clone(&self.shutdown);
        let max_depth = self.config.max_depth;

        let dispatcher = tokio::spawn(async move {
            let mut url_rx = url_rx;
            while let Some((url, depth)) = url_rx.recv().await {
                // 检查是否已访问
                {
                    let mut vis = visited.lock().await;
                    if vis.contains(&url) {
                        continue;
                    }
                    vis.insert(url.clone());
                }

                // 获取信号量许可（限制并发）
                let permit = semaphore.clone().acquire_owned().await.unwrap();

                let client = client.clone();
                let result_tx = result_tx.clone();
                let url_tx = worker_url_tx.clone();
                let visited = Arc::clone(&visited);
                let max_urls = self.config.max_urls;

                tokio::spawn(async move {
                    let _permit = permit;  // 持有许可直到任务完成

                    match timeout(
                        Duration::from_secs(10),
                        client.get(&url).send()
                    ).await {
                        Ok(Ok(resp)) => {
                            let status = resp.status().as_u16();
                            let body = resp.text().await.unwrap_or_default();
                            let links = extract_links(&url, &body);

                            let result = CrawlResult {
                                url: url.clone(),
                                status,
                                body_len: body.len(),
                                links: links.clone(),
                            };
                            let _ = result_tx.send(result).await;

                            // 如果深度允许，把新链接加入队列
                            if depth < max_depth {
                                for link in links {
                                    let vis = visited.lock().await;
                                    if !vis.contains(&link) && vis.len() < max_urls {
                                        drop(vis);
                                        let _ = url_tx.send((link, depth + 1)).await;
                                    }
                                }
                            }
                        }
                        Ok(Err(e)) => eprintln!("请求失败 {}: {}", url, e),
                        Err(_) => eprintln!("超时: {}", url),
                    }
                });
            }
        });

        // 优雅关闭监控
        let shutdown_clone = Arc::clone(&self.shutdown);
        tokio::spawn(async move {
            tokio::signal::ctrl_c().await.unwrap();
            println!("\n⚠️ 收到 Ctrl+C，正在优雅关闭...");
            shutdown_clone.notify_waiters();
        });

        // 收集结果
        drop(result_tx);  // 关闭发送端（dispatcher 持有一份）
        let mut results = vec![];
        while let Some(result) = result_rx.recv().await {
            println!("✅ [{}] {} ({} bytes, {} links)",
                     result.status, result.url, result.body_len, result.links.len());
            results.push(result);
        }

        dispatcher.abort();
        results
    }

    /// 请求优雅关闭
    fn shutdown(&self) {
        self.shutdown.notify_waiters();
    }
}

/// 从 HTML 中提取链接（简化版）
fn extract_links(base_url: &str, html: &str) -> Vec<String> {
    let mut links = Vec::new();
    for line in html.lines() {
        if let Some(start) = line.find("href=\"") {
            if let Some(rest) = line[start + 6..].split('"').next() {
                if rest.starts_with("http") {
                    links.push(rest.to_string());
                } else if rest.starts_with('/') {
                    links.push(format!("{}{}", base_url, rest));
                }
            }
        }
    }
    links.dedup();
    links
}

#[tokio::main]
async fn main() {
    let crawler = Crawler::new(CrawlerConfig {
        max_concurrent: 5,
        max_depth: 2,
        max_urls: 50,
        ..Default::default()
    });

    let results = crawler.crawl(vec![
        "https://example.com".to_string(),
    ]).await;

    println!("\n📊 爬取完成:");
    println!("  总页面数: {}", results.len());
    println!("  总字节数: {}", results.iter().map(|r| r.body_len).sum::<usize>());
    println!("  成功率: {:.1}%",
        results.iter().filter(|r| r.status == 200).count() as f64
        / results.len() as f64 * 100.0
    );
}