深入Rust：Tokio多线程调度架构的原理、实践与性能优化

use tokio::task::Builder;

#[tokio::main]
async fn main() {
    // 提交高优先级任务
    Builder::new()
        .name("user-request-handler")
        .priority(tokio::task::Priority::High)
        .spawn(async {
            println!("处理用户请求（高优先级）");
            // 业务逻辑...
        })
        .unwrap();

    // 提交低优先级任务
    Builder::new()
        .name("log-writer")
        .priority(tokio::task::Priority::Low)
        .spawn(async {
            println!("写入日志（低优先级）");
            // 日志逻辑...
        })
        .unwrap();
}

use tokio::task;
use std::thread;

#[tokio::main(worker_threads = 4)] // 显式设置4个Worker线程（便于观察）
async fn main() {
    println!("主线程ID：{:?}", thread::current().id());
    println!("Worker线程数：4（通过worker_threads配置）\n");

    // 启动10个异步任务
    let mut handles = vec![];
    for i in 0..10 {
        let handle = task::spawn(async move {
            // 打印任务ID和执行线程ID
            let thread_id = thread::current().id();
            println!("任务{} - 执行线程ID：{:?}", i, thread_id);
            // 模拟IO等待（让任务有机会被窃取）
            tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
        });
        handles.push(handle);
    }

    // 等待所有任务完成
    for handle in handles {
        handle.await.unwrap();
    }
}

主线程ID：ThreadId(1)
Worker线程数：4（通过worker_threads配置）

任务0 - 执行线程ID：ThreadId(2)
任务1 - 执行线程ID：ThreadId(3)
任务2 - 执行线程ID：ThreadId(4)
任务3 - 执行线程ID：ThreadId(5)
任务4 - 执行线程ID：ThreadId(2)  // 线程2再次执行任务（本地队列有任务）
任务5 - 执行线程ID：ThreadId(3)  // 线程3再次执行任务
任务6 - 执行线程ID：ThreadId(4)  // 线程4再次执行任务
任务7 - 执行线程ID：ThreadId(5)  // 线程5再次执行任务
任务8 - 执行线程ID：ThreadId(2)  // 线程2空闲后，窃取其他线程的任务
任务9 - 执行线程ID：ThreadId(3)  // 线程3空闲后，窃取其他线程的任务

use tokio::task;
use std::fs;
use std::thread;
use std::time::Instant;

// 模拟阻塞操作：读取大文件（或睡眠2秒）
fn blocking_operation(task_name: &str) {
    println!("{} - 开始阻塞操作（线程ID：{:?}）", task_name, thread::current().id());
    // 模拟阻塞（实际场景可能是fs::read_to_string、CPU密集计算等）
    thread::sleep(std::time::Duration::from_secs(2));
    println!("{} - 阻塞操作完成", task_name);
}

#[tokio::main(worker_threads = 2)] // 仅2个Worker线程（放大阻塞影响）
async fn main() {
    let start = Instant::now();

    // 任务1：直接在Worker中执行阻塞操作（错误做法）
    let task1 = task::spawn(async {
        blocking_operation("任务1（错误：直接阻塞）");
    });

    // 任务2：用spawn_blocking在阻塞线程池执行（正确做法）
    let task2 = task::spawn_blocking(|| {
        blocking_operation("任务2（正确：spawn_blocking）");
    });

    // 任务3：普通异步任务（验证是否被阻塞）
    let task3 = task::spawn(async {
        println!("任务3 - 等待1秒（线程ID：{:?}）", thread::current().id());
        tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
        println!("任务3 - 执行完成");
    });

    // 等待所有任务完成
    task1.await.unwrap();
    task2.await.unwrap();
    task3.await.unwrap();

    println!("总耗时：{:?}", start.elapsed());
}

任务1（错误：直接阻塞） - 开始阻塞操作（线程ID：ThreadId(2)）  // Worker线程2被阻塞
任务2（正确：spawn_blocking） - 开始阻塞操作（线程ID：ThreadId(4)）  // 阻塞线程池的线程4
任务3 - 等待1秒（线程ID：ThreadId(3)）  // 仅有的另一个Worker线程3
任务3 - 执行完成（1秒后）
任务1（错误：直接阻塞） - 阻塞操作完成（2秒后，Worker线程2释放）
任务2（正确：spawn_blocking） - 阻塞操作完成（2秒后，阻塞线程4释放）
总耗时：2.003s

use tokio::runtime::Builder;
use tokio::task;
use std::time::Instant;

// 模拟IO密集型任务（大部分时间在等待IO）
async fn io_bound_task(id: u32) {
    println!("IO任务{} - 等待IO（线程ID：{:?}）", id, std::thread::current().id());
    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await; // 模拟IO等待
    println!("IO任务{} - 执行完成", id);
}

// 模拟CPU密集型任务（大部分时间在计算）
fn cpu_bound_task(id: u32) {
    println!("CPU任务{} - 开始计算（线程ID：{:?}）", id, std::thread::current().id());
    let mut sum = 0;
    for i in 0..1_000_000_000 { // 模拟CPU密集计算
        sum += i;
    }
    println!("CPU任务{} - 计算完成（sum={}）", id, sum);
}

fn main() {
    // 1. 配置IO密集型业务的Runtime：Worker线程数=CPU核心数（默认），阻塞线程池最大=10
    let io_runtime = Builder::new_multi_thread()
        .worker_threads(num_cpus::get()) // Worker线程数=CPU核心数（充分利用多核处理IO任务）
        .max_blocking_threads(10)        // 阻塞线程池最大10个（IO密集场景阻塞任务少）
        .thread_name("io-worker")        // 线程命名（便于日志排查）
        .build()
        .unwrap();

    // 运行IO密集型任务
    let io_start = Instant::now();
    io_runtime.block_on(async {
        let mut handles = vec![];
        for i in 0..20 { // 启动20个IO任务（远多于Worker线程数）
            handles.push(task::spawn(io_bound_task(i)));
        }
        for handle in handles {
            handle.await.unwrap();
        }
    });
    println!("IO密集任务总耗时：{:?}\n", io_start.elapsed());

    // 2. 配置CPU密集型业务的Runtime：Worker线程数=CPU核心数（避免上下文切换），阻塞线程池最大=20
    let cpu_runtime = Builder::new_multi_thread()
        .worker_threads(num_cpus::get()) // CPU密集任务并行数=CPU核心数（减少切换开销）
        .max_blocking_threads(20)        // 阻塞线程池最大20个（CPU密集场景可能有更多阻塞任务）
        .thread_name("cpu-worker")
        .build()
        .unwrap();

    // 运行CPU密集型任务（用spawn_blocking提交）
    let cpu_start = Instant::now();
    cpu_runtime.block_on(async {
        let mut handles = vec![];
        for i in 0..num_cpus::get() * 2 { // 启动2倍CPU核心数的CPU任务
            let handle = task::spawn_blocking(move || cpu_bound_task(i));
            handles.push(handle);
        }
        for handle in handles {
            handle.await.unwrap();
        }
    });
    println!("CPU密集任务总耗时：{:?}", cpu_start.elapsed());
}

// IO密集任务（20个任务，8个Worker线程）
IO任务0 - 等待IO（线程ID：ThreadId(2)）
...（20个IO任务分散在8个Worker线程）
IO任务0 - 执行完成
...（20个IO任务在100ms内完成）
IO密集任务总耗时：102ms  // 所有任务并行等待IO，耗时≈100ms

// CPU密集任务（16个任务，8个Worker线程，阻塞线程池最大20）
CPU任务0 - 开始计算（线程ID：ThreadId(10)）
...（16个CPU任务分散在8个阻塞线程）
CPU任务0 - 计算完成（sum=499999999500000000）
...（16个任务分两批执行，每批8个，总耗时≈2倍单任务时间）
CPU密集任务总耗时：2.1s  // 单任务耗时≈1s，两批并行，总耗时≈2s

use tokio::task;
use std::rc::Rc;

#[tokio::main]
async fn main() {
    let rc = Rc::new(5); // Rc是非Send类型
    // 错误：`Rc<i32>` cannot be sent between threads safely
    task::spawn(async move {
        println!("rc: {}", rc);
    }).await.unwrap();
}

use tokio::task::{LocalSet, spawn_local};
use std::rc::Rc;

#[tokio::main]
async fn main() {
    // 创建LocalSet，用于执行非Send任务
    let local_set = LocalSet::new();

    local_set.run_until(async {
        let rc = Rc::new(5);
        // 用spawn_local提交非Send任务（仅在当前Worker线程执行）
        spawn_local(async move {
            println!("rc: {}", rc); // 正确：无跨线程调度
        }).await.unwrap();
    }).await;
}

use tokio::task;
use std::thread;

#[tokio::main(worker_threads = 1)] // 仅1个Worker线程
async fn main() {
    // 错误：长时间阻塞用block_in_place，导致Worker线程被占用，其他任务无法执行
    task::spawn(async {
        task::block_in_place(|| {
            thread::sleep(std::time::Duration::from_secs(2)); // 长时间阻塞
        });
    }).await.unwrap();

    // 该任务会被阻塞2秒，直到block_in_place完成
    task::spawn(async {
        println!("这个任务会延迟2秒执行");
    }).await.unwrap();
}

use tokio::task;
use std::thread;

#[tokio::main(worker_threads = 1)]
async fn main() {
    // 正确：长时间阻塞用spawn_blocking，不占用Worker线程
    task::spawn_blocking(|| {
        thread::sleep(std::time::Duration::from_secs(2));
    }).await.unwrap();

    // 该任务会立即执行，无需等待
    task::spawn(async {
        println!("这个任务会立即执行");
    }).await.unwrap();
}