上一篇的泛型仓库能替换存储实现但所有调用仍在单线程顺序执行。本篇先用线程与通道完成 CPU/阻塞工作再用 Tokio 演示异步调度解释Send、Sync、锁范围和取消安全而不是只记住.await。一、痛点并发、并行与异步不是同义词并发描述多个任务在时间上推进并行表示多个任务在多个核心同时执行异步是一种在等待时主动让出执行权的并发模型。CPU 密集计算适合线程池并行成千上万网络连接大多在等待 I/O适合异步运行时。把阻塞文件或计算直接放进异步执行器会卡住承载其他任务的工作线程。Rust 的类型系统把部分并发错误提前Send表示值可在线程间转移Sync表示共享引用可在线程间安全使用。它们多由编译器自动推导。Rc非原子计数不可跨线程Arc使用原子计数可共享所有权。共享可变状态还需Mutex或RwLock计数本身不会保护内部数据。优先消息传递生产者移动消息给消费者所有权边界简单。确实需要多个线程读写同一映射时才用ArcMutexT并尽量缩短持锁时间。二、原理线程所有权与通道背压thread::spawn闭包通常要求static用move把拥有值移入线程。通道发送端可克隆给多个生产者接收端迭代直到所有发送端释放。无界通道可能在生产快于消费时增长内存真实系统应采用有界队列或明确限流。usestd::sync::mpsc;usestd::thread;#[derive(Debug)]structResultItem{id:u32,score:u32}fnscore(id:u32)-u32{(1..id).map(|value|value*value).sum()}fnmain(){let(sender,receiver)mpsc::channel();letmuthandlesVec::new();foridin1..4{letworker_sendersender.clone();handles.push(thread::spawn(move||{letitemResultItem{id,score:score(id)};worker_sender.send(item).expect(接收端仍存在);}));}drop(sender);forhandleinhandles{handle.join().expect(工作线程不应 panic);}letmutresults:Vec_receiver.into_iter().collect();results.sort_by_key(|item|item.id);foriteminresults{println!(id{} score{},item.id,item.score);}}运行输出id1 score1 id2 score5 id3 score14 id4 score30线程完成顺序不确定所以输出前按编号排序。并发测试若依赖自然调度顺序会偶发失败稳定输出必须来自显式排序或协议保证。join不仅等待还传播线程 panic生产代码应转换为错误并决定是否继续。三、实现用 Tokio 调度等待型任务异步函数调用后返回Future只有被.await或运行时轮询才推进。等待点可能暂停当前任务因此不要跨.await长时间持有同步锁。下面项目在Cargo.toml加tokio { version 1, features [full] }每个任务模拟一次远程请求。usetokio::task::JoinSet;usetokio::time::{sleep,Duration};#[derive(Debug)]structTask{id:u32,title:staticstr}asyncfnfetch(task:Task)-String{letdelayDuration::from_millis((5-task.id)asu64*10);sleep(delay).await;format!({}:{},task.id,task.title)}#[tokio::main]asyncfnmain(){lettasks[Task{id:1,title:文档},Task{id:2,title:测试},Task{id:3,title:服务},Task{id:4,title:部署},];letmutsetJoinSet::new();fortaskintasks{set.spawn(fetch(task));}letmutvaluesVec::new();whileletSome(result)set.join_next().await{values.push(result.expect(任务不应 panic));}values.sort();forvalueinvalues{println!(result{value});}}运行输出result1:文档 result2:测试 result3:服务 result4:部署JoinSet管理一组动态任务并显式接收完成结果。直接spawn后丢弃句柄会让错误无处观察。超时可用tokio::time::timeout包裹 future但超时只停止等待底层操作是否取消、是否产生部分副作用要查具体 API 契约。四、踩坑锁、阻塞与无限制 spawn持MutexGuard做网络请求会让其他任务等待最慢 I/O。正确做法是在锁内读取或更新最小数据释放锁后再.await。标准库锁适合短且不跨等待点的临界区确需跨异步等待的锁可用 Tokio 版本但更应重新审视状态所有权。异步不自动更快。少量顺序任务使用普通阻塞代码更简单CPU 密集计算应交给专用线程池或spawn_blocking否则执行器延迟上升。无限tokio::spawn会造成连接、内存和下游压力应用信号量限制在途数量并让队列长度可观测。死锁不仅是两把锁互等也可能因任务持锁等待另一个需要同锁的任务。统一锁顺序、减少共享状态、用消息循环单独拥有状态都能降低风险。并发 bug 难靠一次运行发现设计时要明确关闭、取消、超时和错误传播路径。五、验证让非确定调度产生确定结论测试结果集合而非完成顺序时间相关测试使用 Tokio 的暂停时间能力减少真实睡眠。通道必须验证发送端关闭后消费者能退出有界队列要测试满载策略。共享计数测试应多次运行并用线程检测工具或模型检查库补充关键同步逻辑。现在任务可以并发处理并异步等待但依赖和特性还写在单一包里。下一篇转向 Cargo 工程化拆分库与二进制、使用 feature、workspace、锁文件和发布配置让后面的网络服务拥有清晰项目骨架。参考来源Rust BookFearless ConcurrencyRust 标准库std::syncTokio TutorialAsync Book异步编程 觉得有用就点个赞 收藏方便回头查阅有疑问直接在评论区留言我看到都会回。 本文属于《Rust 入门实战》系列持续更新关注不迷路。 文章里的代码都能直接跑。想要可直接 clone 的完整工程 配套部署脚本 / 踩坑清单评论一声或发邮件到cj2664qq.com我免费发你。如果你正好在做类似系统、或有工程化难题想找人做也欢迎邮件聊一句——我按实际情况评估能落地的就接单或出方案。评论和邮件都能直接找到我不用跳别的平台。
上一篇的泛型仓库能替换存储实现但所有调用仍在单线程顺序执行。本篇先用线程与通道完成 CPU/阻塞工作再用 Tokio 演示异步调度解释Send、Sync、锁范围和取消安全而不是只记住.await。一、痛点并发、并行与异步不是同义词并发描述多个任务在时间上推进并行表示多个任务在多个核心同时执行异步是一种在等待时主动让出执行权的并发模型。CPU 密集计算适合线程池并行成千上万网络连接大多在等待 I/O适合异步运行时。把阻塞文件或计算直接放进异步执行器会卡住承载其他任务的工作线程。Rust 的类型系统把部分并发错误提前Send表示值可在线程间转移Sync表示共享引用可在线程间安全使用。它们多由编译器自动推导。Rc非原子计数不可跨线程Arc使用原子计数可共享所有权。共享可变状态还需Mutex或RwLock计数本身不会保护内部数据。优先消息传递生产者移动消息给消费者所有权边界简单。确实需要多个线程读写同一映射时才用ArcMutexT并尽量缩短持锁时间。二、原理线程所有权与通道背压thread::spawn闭包通常要求static用move把拥有值移入线程。通道发送端可克隆给多个生产者接收端迭代直到所有发送端释放。无界通道可能在生产快于消费时增长内存真实系统应采用有界队列或明确限流。usestd::sync::mpsc;usestd::thread;#[derive(Debug)]structResultItem{id:u32,score:u32}fnscore(id:u32)-u32{(1..id).map(|value|value*value).sum()}fnmain(){let(sender,receiver)mpsc::channel();letmuthandlesVec::new();foridin1..4{letworker_sendersender.clone();handles.push(thread::spawn(move||{letitemResultItem{id,score:score(id)};worker_sender.send(item).expect(接收端仍存在);}));}drop(sender);forhandleinhandles{handle.join().expect(工作线程不应 panic);}letmutresults:Vec_receiver.into_iter().collect();results.sort_by_key(|item|item.id);foriteminresults{println!(id{} score{},item.id,item.score);}}运行输出id1 score1 id2 score5 id3 score14 id4 score30线程完成顺序不确定所以输出前按编号排序。并发测试若依赖自然调度顺序会偶发失败稳定输出必须来自显式排序或协议保证。join不仅等待还传播线程 panic生产代码应转换为错误并决定是否继续。三、实现用 Tokio 调度等待型任务异步函数调用后返回Future只有被.await或运行时轮询才推进。等待点可能暂停当前任务因此不要跨.await长时间持有同步锁。下面项目在Cargo.toml加tokio { version 1, features [full] }每个任务模拟一次远程请求。usetokio::task::JoinSet;usetokio::time::{sleep,Duration};#[derive(Debug)]structTask{id:u32,title:staticstr}asyncfnfetch(task:Task)-String{letdelayDuration::from_millis((5-task.id)asu64*10);sleep(delay).await;format!({}:{},task.id,task.title)}#[tokio::main]asyncfnmain(){lettasks[Task{id:1,title:文档},Task{id:2,title:测试},Task{id:3,title:服务},Task{id:4,title:部署},];letmutsetJoinSet::new();fortaskintasks{set.spawn(fetch(task));}letmutvaluesVec::new();whileletSome(result)set.join_next().await{values.push(result.expect(任务不应 panic));}values.sort();forvalueinvalues{println!(result{value});}}运行输出result1:文档 result2:测试 result3:服务 result4:部署JoinSet管理一组动态任务并显式接收完成结果。直接spawn后丢弃句柄会让错误无处观察。超时可用tokio::time::timeout包裹 future但超时只停止等待底层操作是否取消、是否产生部分副作用要查具体 API 契约。四、踩坑锁、阻塞与无限制 spawn持MutexGuard做网络请求会让其他任务等待最慢 I/O。正确做法是在锁内读取或更新最小数据释放锁后再.await。标准库锁适合短且不跨等待点的临界区确需跨异步等待的锁可用 Tokio 版本但更应重新审视状态所有权。异步不自动更快。少量顺序任务使用普通阻塞代码更简单CPU 密集计算应交给专用线程池或spawn_blocking否则执行器延迟上升。无限tokio::spawn会造成连接、内存和下游压力应用信号量限制在途数量并让队列长度可观测。死锁不仅是两把锁互等也可能因任务持锁等待另一个需要同锁的任务。统一锁顺序、减少共享状态、用消息循环单独拥有状态都能降低风险。并发 bug 难靠一次运行发现设计时要明确关闭、取消、超时和错误传播路径。五、验证让非确定调度产生确定结论测试结果集合而非完成顺序时间相关测试使用 Tokio 的暂停时间能力减少真实睡眠。通道必须验证发送端关闭后消费者能退出有界队列要测试满载策略。共享计数测试应多次运行并用线程检测工具或模型检查库补充关键同步逻辑。现在任务可以并发处理并异步等待但依赖和特性还写在单一包里。下一篇转向 Cargo 工程化拆分库与二进制、使用 feature、workspace、锁文件和发布配置让后面的网络服务拥有清晰项目骨架。参考来源Rust BookFearless ConcurrencyRust 标准库std::syncTokio TutorialAsync Book异步编程 觉得有用就点个赞 收藏方便回头查阅有疑问直接在评论区留言我看到都会回。 本文属于《Rust 入门实战》系列持续更新关注不迷路。 文章里的代码都能直接跑。想要可直接 clone 的完整工程 配套部署脚本 / 踩坑清单评论一声或发邮件到cj2664qq.com我免费发你。如果你正好在做类似系统、或有工程化难题想找人做也欢迎邮件聊一句——我按实际情况评估能落地的就接单或出方案。评论和邮件都能直接找到我不用跳别的平台。
