教程文档【免费下载链接】bookThe Rust Programming Language项目地址https://gitcode.com/gh_mirrors/bo/book点击查看免费下载本篇以《The Rust Programming Language》本仓库即该书的开源仓库第 17 章第 2 节 src/ch17-02-concurrency-with-async.md 为主线讲解如何用 async/await 与trpl教学运行时解决第 16 章用线程解决过的并发问题任务创建、任务汇合、消息传递、多生产者广播与所有权转移。读完你将掌握spawn_task、join、join!、async channel、while let接收循环以及async move的核心用法并能理解 async 与线程在调度与公平性上的本质差异。从线程到 Future同一批并发问题的另一种解法第 16 章我们用std::thread处理过的三个典型并发场景——双任务计数、线程间消息传递、等待所有线程结束——在本节全部用 async 重新实现了一遍。两者在很多 API 上看起来非常相似但行为与性能特性几乎总是不一样线程由操作系统调度async 任务由**运行时runtime**调度线程是重量级资源每个线程都有独立的栈与系统资源开销而 async 任务/Future是轻量级的一个线程上可以并发运行成百上千个任务线程间的同步等待用join()阻塞async 中的等待用await让出控制权不阻塞。由于 API 相似但语义不同本节反复强调即使写法长得像也不要假设行为相同。下面我们按文档的推进顺序逐一展开所有代码均来自仓库 listings/ch17-async-await/ 中可运行的示例。用spawn_task创建新任务顶层入口trpl::block_on在开始之前先交代所有示例的“外壳”。main函数本身不能是异步的因此我们用它包一层extern crate trpl; // required for mdbook test use std::time::Duration; fn main() { trpl::block_on(async { // 你的异步代码写在这里 }); }从源码看trpl::block_on的实现非常直白packages/trpl/src/lib.rspub fn block_onF: Future(future: F) - F::Output { let rt Runtime::new().unwrap(); rt.block_on(future) }它每次调用都会新建一个 TokioRuntime实例然后运行给定的 future 直到完成。这正是block_on的设计意图让你自己选择在哪里阻塞把同步代码与异步代码衔接起来block_on内部的代码不会阻塞线程但调用block_on的同步代码会阻塞到 future 完成。从本章往后src/ch17-01-futures-and-syntax.md 至 src/ch17-06-futures-tasks-threads.md所有示例都包含这层trpl::block_on包裹代码文中常省略实际编写时务必带上。两个循环在单任务与主任务间并发第 16 章 “用spawn创建新线程” 的第一个练习是在两条线程上分别计数这里用 async 等价重写仓库示例 listing-17-06/src/main.rsuse std::time::Duration; fn main() { trpl::block_on(async { trpl::spawn_task(async { for i in 1..10 { println!(hi number {i} from the first task!); trpl::sleep(Duration::from_millis(500)).await; } }); for i in 1..5 { println!(hi number {i} from the second task!); trpl::sleep(Duration::from_millis(500)).await; } }); }trpl::spawn_task与thread::spawn的 API 外观很像源码中它直接再导出tokio::task::spawn见 packages/trpl/src/lib.rstrpl::sleep则是thread::sleep的异步版本再导出tokio::time::sleep。关键差异在两处spawn_task接收的是一个future这里是async块而不是一个闭包trpl::sleep(...)之后必须跟.await让运行时在等待的 500 毫秒里可以去推进别的任务。运行结果与线程版相似且消息顺序不固定——因为并发任务推进的次序取决于运行时调度hi number 1 from the second task! hi number 1 from the first task! hi number 2 from the first task! hi number 2 from the second task! hi number 3 from the first task! hi number 3 from the second task! hi number 4 from the first task! hi number 4 from the second task! hi number 5 from the first task!注意这个版本在main里的for循环结束后就立刻终止了——spawn_task生成的任务会在main结束、block_on返回时被一并关停即使第一个任务本应再打印 5 次。要让任务跑完全程必须用join handle等它完成。用 join handle 等任务跑完线程版用join()阻塞等待线程结束在 async 世界里join handle 本身就是 future直接await即可listing-17-07/src/main.rslet handle trpl::spawn_task(async { for i in 1..10 { println!(hi number {i} from the first task!); trpl::sleep(Duration::from_millis(500)).await; } }); for i in 1..5 { println!(hi number {i} from the second task!); trpl::sleep(Duration::from_millis(500)).await; } handle.await.unwrap();handle的Output类型是Result所以 await 之后需要unwrap()。此时两个循环都完整跑完输出会延续到hi number 9 from the first task!。到这里读者可能会想既然任务不需要额外系统线程那是不是连spawn_task都可以省掉答案是肯定的——async 块本身编译成匿名 future直接构造两个 future再用trpl::join让运行时把它们都跑完即可。用trpl::join公平地汇合两个 future第 16 章的 “等待所有线程结束” 使用了JoinHandle::jointrpl::join是它的 future 版接收两个 future返回一个新的 future其Output是包含两者输出的元组且要等两者都完成后才结束listing-17-08/src/main.rslet fut1 async { for i in 1..10 { println!(hi number {i} from the first task!); trpl::sleep(Duration::from_millis(500)).await; } }; let fut2 async { for i in 1..5 { println!(hi number {i} from the second task!); trpl::sleep(Duration::from_millis(500)).await; } }; trpl::join(fut1, fut2).await;这里我们 await 的是trpl::join返回的新 future而不是分别 awaitfut1、fut2那会退化成顺序执行。trpl::join的返回值只是两个单元值组成的元组直接忽略。运行输出最大的特点是每次顺序完全一致hi number 1 from the first task! hi number 1 from the second task! hi number 2 from the first task! hi number 2 from the second task! hi number 3 from the first task! hi number 3 from the second task! hi number 4 from the first task! hi number 4 from the second task! hi number 5 from the first task! hi number 6 from the first task! hi number 7 from the first task! hi number 8 from the first task! hi number 9 from the first task!这与线程版、spawn_task版截然不同。原因在于trpl::join是“公平”fair的它会以相同的频率轮询每个 future交替推进只要另一个 future 就绪就不让某一个抢先跑远。而线程的推进完全由操作系统决定async 的推进则由运行时决定实践中运行时底层可能仍会借用系统线程来管理并发因此保证公平性对运行时而言成本更高但并非不可能。需要注意的是运行时并不承诺对任何操作都保证公平很多运行时还会提供不同 API 让你自行选择是否要公平。动手实验理解 future 何时被轮询文档给出三组值得亲自验证的变体改完先预测输出再运行对照去掉其中一个或两个循环外面的async块会变成同步执行trpl::sleep(...).await无法脱离 async 上下文需配合改法理解定义完每个 async 块后立刻await它先跑完fut1再跑fut2退化为顺序执行只把第一个循环包进 async 块第二个循环放在块外在第二个循环体之后才 await 第一个 future前半段顺序、后半段并发。这三种改法分别演示了async 块是并发的基本单元、await 的位置决定执行次序、以及并发只发生在 future 真正被交给运行时共同推进的时候。用异步 channel 在任务间传递数据创建 channelasync 版与线程版的三个区别接下来用消息传递在 futures 之间共享数据。先看最小的单块版本listing-17-09/src/main.rsfn main() { trpl::block_on(async { let (tx, mut rx) trpl::channel(); let val String::from(hi); tx.send(val).unwrap(); let received rx.recv().await.unwrap(); println!(received {received}); }); }trpl::channel是第 16 章std::sync::mpsc::channel的 async 版同样是多生产者单消费者MPSC。从源码看packages/trpl/src/lib.rstrpl::channel实际再导出的是tokio::sync::mpsc::unbounded_channel并把UnboundedReceiver命名为Receiver、UnboundedSender命名为Sender——教科书为了教学简洁抹平了 Tokio 的channel/unbounded_channel命名差异让读者聚焦 sync 与 async API 之间更重要的区别。与线程版相比async 版的 API 只有三处不同对比项线程版std::sync::mpscasync 版trpl::channel接收端不可变rx可变mut rxrecv直接返回值阻塞等待返回future需要.awaitsend可能阻塞有界/同步语义不阻塞无界 channel同步的Receiver::recv会阻塞直到收到消息trpl::Receiver::recv不会阻塞——它把控制权交回运行时直到收到一条消息或发送端关闭。send之所以不需要 await是因为我们用的是无界unboundedchannel发送永远立即可完成。再注意两点第一消息立刻到达第二这个例子还没有任何并发——所有语句在同一个 async 块内顺序执行和不用 future 完全一样。并发要等我们把收发两边拆进各自的 async 块才出现。发送多条消息并配合 sleep暴露第一个问题为了展示真正的收发节奏文档发送一串消息并在每两条之间 sleep 500 毫秒listing-17-10/src/main.rslet (tx, mut rx) trpl::channel(); let vals vec![ String::from(hi), String::from(from), String::from(the), String::from(future), ]; for val in vals { tx.send(val).unwrap(); trpl::sleep(Duration::from_millis(500)).await; } while let Some(value) rx.recv().await { println!(received {value}); }因为不知道会来多少条消息这里不能手写四次rx.recv().await而是用了while let条件循环if let的循环版本见第 6 章 “用if let精简控制流”只要rx.recv().await解析为Some(message)就继续执行循环体当 channel 关闭无论此前有没有消息到达它解析为None循环结束——也就是停止轮询。每次循环体结束都会再次命中 await 点运行时暂停该 future 直到下一条消息到达。这个版本暴露了两个问题消息不是每 500 毫秒来一条而是在程序启动 2 秒后一次性全部到达程序永不退出必须用Ctrl-C强制终止。第二个问题我们先放一放先解决第一个——它引出一个重要的心智模型。单个 async 块内的代码是线性执行的问题一的根因在一个 async 块内部await关键字出现的顺序就是它被执行的顺序。Listing 17-10 只有一个 async 块所以它内部完全是线性的先完成所有tx.send穿插所有trpl::sleep及各自的 await 点然后while let循环才轮到recv的 await 点。整个过程中根本没有并发睡眠的 2 秒全部发生在接收之前。要让每条消息之间都有 500 毫秒的间隔必须把tx的发送逻辑和rx的接收逻辑拆进两个独立的 async 块再用trpl::join让运行时分别推进它们listing-17-11/src/main.rslet (tx, mut rx) trpl::channel(); let tx_fut async { let vals vec![ String::from(hi), String::from(from), String::from(the), String::from(future), ]; for val in vals { tx.send(val).unwrap(); trpl::sleep(Duration::from_millis(500)).await; } }; let rx_fut async { while let Some(value) rx.recv().await { println!(received {value}); } }; trpl::join(tx_fut, rx_fut).await;与 Listing 17-8 同理这里 await 的是trpl::join的结果而不是两个 future 本身若按顺序分别 await又会退化成顺序流。改完后消息果然以 500 毫秒的间隔陆续打印。用async move转移所有权来修复无限循环但程序仍然不退出。文档给出了完整的原因链条trpl::join返回的 future 只有在其两个 future都完成时才完成tx_fut在发完vals里最后一条消息并睡完觉后完成rx_fut要等while let循环结束才完成while let循环要等rx.recv().await返回None才结束rx.recv只在 channel 另一端关闭时返回Nonechannel 只在调用rx.close或发送端tx被 drop 时关闭代码里既没调用rx.closetx又直到传给block_on的最外层 async 块结束才会被 drop而这个块正被trpl::join阻塞着无法结束——于是回到第 1 条形成死循环。破局的关键在第 6 条目前发送消息的 async 块只是借用txsend 不需要所有权如果改为把tx移动进该 async 块块结束时tx就被 dropchannel 随之关闭rx.recv返回None一切自然收尾。这正是async move的用途——move关键字对 async 块的作用与对闭包相同详见第 13 章 “捕获引用或转移所有权” 与第 16 章 “在线程中使用move闭包”。listing-17-12/src/main.rs 只需把发送块从async改成async movelet tx_fut async move { let vals vec![ String::from(hi), String::from(from), String::from(the), String::from(future), ]; for val in vals { tx.send(val).unwrap(); trpl::sleep(Duration::from_millis(500)).await; } }; let rx_fut async { while let Some(value) rx.recv().await { println!(received {value}); } }; trpl::join(tx_fut, rx_fut).await;运行这一版最后一条消息收发完毕后程序优雅退出。用join!宏汇合任意数量的 futureasync channel 支持多生产者因此可以对tx调用clone从多个 future 发送消息listing-17-13/src/main.rslet (tx, mut rx) trpl::channel(); let tx1 tx.clone(); let tx1_fut async move { let vals vec![ String::from(hi), String::from(from), String::from(the), String::from(future), ]; for val in vals { tx1.send(val).unwrap(); trpl::sleep(Duration::from_millis(500)).await; } }; let rx_fut async { while let Some(value) rx.recv().await { println!(received {value}); } }; let tx_fut async move { let vals vec![ String::from(more), String::from(messages), String::from(for), String::from(you), ]; for val in vals { tx.send(val).unwrap(); trpl::sleep(Duration::from_millis(1500)).await; } }; trpl::join!(tx1_fut, tx_fut, rx_fut);要点有三先clone再转移在第一个 async 块外let tx1 tx.clone()随后把tx1移入第一个块、把原tx移入新块。新块放在接收块之后完全没问题——关键是被 await 的顺序而不是被创建的顺序。两个发送块都必须是async move否则tx和tx1都不会在块结束时 drop又会回到前面那个无限循环。从trpl::join换成trpl::join!join!宏可以同时 await 编译期已知数量的任意多个 future源码中join与join!均由futurescrate 再导出见 packages/trpl/src/lib.rs。若 future 数量在运行期才知道则需要用流streams来处理那是本章后续小节的内容src/ch17-04-streams.md。两个发送端分别以 500ms 与 1500ms 的间隔发送接收端便以对应的不同节奏打印received hi received more received from received the received messages received future received for received you小结与下一步本节完整走完了四个并发能力trpl::spawn_task生成独立 async 任务配合 join handle 的await等待其完成trpl::join公平地同时推进两个 future直到两者都完成异步 channeltrpl::channel的无界 MPSC 语义、recv().await的非阻塞行为以及用while let循环接收未知数量的消息async move与join!通过转移所有权关闭 channel 实现优雅退出通过克隆发送端与join!宏汇合多个 future。贯穿始终的核心差异是线程由操作系统抢占式调度async 任务由运行时协作式调度。协作式调度意味着每个 future 只有在命中await点时才让出控制权因此await出现的位置直接决定执行次序而trpl::join的公平轮询策略又让多个 future 得以交错推进。下一步文档将讨论如何主动把控制权交还给运行时——即显式让出yield机制以及更复杂的 future 组合方式见 src/ch17-03-more-futures.md。相关仓库资源章节正文src/ch17-02-concurrency-with-async.md、章节导读 src/ch17-00-async-await.md全部可运行示例listings/ch17-async-await/本例对应listing-17-06至listing-17-13trpl教学运行时源码packages/trpl/src/lib.rs其中block_on的实现位于第 65–68 行channel/spawn_task/sleep/join等 API 的再导出位于第 23–47 行前后承接章节src/ch16-01-threads.md、src/ch16-02-message-passing.md、src/ch17-03-more-futures.md如需亲手运行这些示例可在仓库 listings/ch17-async-await/ 下任意示例目录执行cargo run示例已声明trpl依赖见各目录下的Cargo.toml。赞分享教程文档【免费下载链接】bookThe Rust Programming Language项目地址https://gitcode.com/gh_mirrors/bo/book点击查看免费下载相关推荐ComfyUI-WanVideoWrapper完整指南从零开始掌握AI视频生成ComfyUI WanVideoWrapper完整指南从零开始掌握AI视频生成 你是否曾梦想过用AI轻松制作专业级视频却苦于复杂的安装配置和技术门槛Com人工智能大模型媒体生成trpl crate 详解The Rust Programming Language 官方 async 教学支撑库的设计、API 与使用方式trpl crate 详解The Rust Programming Language 官方 async 教学支撑库的设计、API 与使用方式 trpl 是《T教程文档Rust 并发收官Futures、Tasks 与 Threads 如何协同工作The Rust Programming Language 实战解析Rust 并发收官Futures、Tasks 与 Threads 如何协同工作The Rust Programming Language 实战解析 导读教程文档创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考