高級異步編程
在 Rust 精通篇中，我們將深入探索 Rust 的高級異步編程技術。Rust 的異步編程模型基于 Future 特征和異步運行時，提供了高效的非阻塞 I/O 和并發處理能力。在本章中，我們將超越基礎知識，探索如何構建高性能異步系統和自定義執行器。
異步編程回顧
在深入高級主題之前，讓我們簡要回顧 Rust 的異步編程模型：
use std::time::Duration;
use tokio::time;#[tokio::main]
async fn main() {// 創建兩個異步任務let task1 = async {time::sleep(Duration::from_millis(100)).await;println!("任務 1 完成");1};let task2 = async {time::sleep(Duration::from_millis(50)).await;println!("任務 2 完成");2};// 并發執行兩個任務let (result1, result2) = tokio::join!(task1, task2);println!("結果: {} + {} = {}", result1, result2, result1 + result2);
}
Rust 的異步編程基于以下核心概念：
- Future 特征：表示可能尚未完成的計算
- async/await 語法：簡化異步代碼的編寫
- 異步運行時：如 Tokio、async-std 等，負責執行和調度異步任務
- 任務（Task）：可獨立調度的異步執行單元
Future 深入理解
Future 特征的內部機制
Future 特征是 Rust 異步編程的核心：
pub trait Future {type Output;fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>;
}
poll 方法是 Future 的核心，它嘗試將異步計算推進到完成狀態：
- 如果 Future 已完成，返回
Poll::Ready(result) - 如果 Future 尚未完成，返回
Poll::Pending并安排在事件發生時重新調用poll
手動實現 Future
下面是一個簡單的 Future 實現示例：
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::{Duration, Instant};struct Delay {when: Instant,
}impl Future for Delay {type Output = ();fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {if Instant::now() >= self.when {println!("Future 已完成");Poll::Ready(())} else {// 安排在未來某個時刻重新調用 polllet waker = cx.waker().clone();let when = self.when;std::thread::spawn(move || {let now = Instant::now();if now < when {std::thread::sleep(when - now);}waker.wake();});println!("Future 尚未完成");Poll::Pending}}
}#[tokio::main]
async fn main() {let future = Delay {when: Instant::now() + Duration::from_secs(1),};println!("等待 Future 完成...");future.await;println!("主函數結束");
}
Pin 和 Unpin
Pin 類型在異步 Rust 中至關重要，它防止自引用結構在內存中被移動：
use std::marker::PhantomPinned;
use std::pin::Pin;// 自引用結構體
struct SelfReferential {data: String,// 指向 data 字段的指針ptr_to_data: *const String,// 標記此類型不能安全地實現 Unpin_marker: PhantomPinned,
}impl SelfReferential {fn new(data: String) -> Pin<Box<Self>> {let mut boxed = Box::new(SelfReferential {data,ptr_to_data: std::ptr::null(),_marker: PhantomPinned,});let ptr = &boxed.data as *const String;boxed.ptr_to_data = ptr;// 將 Box 轉換為 Pin<Box>Pin::new(boxed)}fn get_data(self: Pin<&Self>) -> &str {// 安全：數據不會被移動，因為它被固定了let self_ref = unsafe { self.get_ref() };&self_ref.data}fn get_ptr(self: Pin<&Self>) -> *const String {self.ptr_to_data}
}fn main() {let pinned = SelfReferential::new("hello".to_string());// 驗證指針確實指向數據let data_ptr = &pinned.data as *const String;let ptr = pinned.as_ref().get_ptr();println!("數據指針: {:?}", data_ptr);println!("存儲的指針: {:?}", ptr);println!("數據: {}", pinned.as_ref().get_data());assert_eq!(data_ptr, ptr);
}
異步運行時深入剖析
執行器（Executor）工作原理
異步執行器負責調度和運行 Future，下面是一個簡單執行器的實現：
use futures::future::BoxFuture;
use futures::task::{waker_ref, ArcWake};
use futures::Future;
use std::sync::{Arc, Mutex};
use std::sync::mpsc::{channel, Sender, Receiver};
use std::task::{Context, Poll};// 任務結構，包含一個 Future
struct Task {future: Mutex<Option<BoxFuture<'static, ()>>>,sender: Sender<Arc<Task>>,
}impl ArcWake for Task {fn wake_by_ref(arc_self: &Arc<Self>) {// 將自己發送到任務隊列，以便重新執行let cloned = arc_self.clone();arc_self.sender.send(cloned).expect("任務隊列已滿");}
}// 簡單的執行器
struct Executor {sender: Sender<Arc<Task>>,receiver: Receiver<Arc<Task>>,
}impl Executor {fn new() -> Self {let (sender, receiver) = channel();Executor { sender, receiver }}// 生成新任務fn spawn(&self, future: impl Future<Output = ()> + 'static + Send) {let task = Arc::new(Task {future: Mutex::new(Some(Box::pin(future))),sender: self.sender.clone(),});self.sender.send(task).expect("任務隊列已滿");}// 運行執行器fn run(&self) {while let Ok(task) = self.receiver.recv() {// 創建 waker 和上下文let waker = waker_ref(&task);let mut context = Context::from_waker(&waker);// 嘗試推進 Futurelet mut future_slot = task.future.lock().unwrap();if let Some(mut future) = future_slot.take() {match Future::poll(Pin::new(&mut future), &mut context) {Poll::Pending => {// Future 尚未完成，放回任務中*future_slot = Some(future);}Poll::Ready(()) => {// Future 已完成，丟棄它// 不需要放回 future_slot}}}}}
}fn main() {let executor = Executor::new();// 生成一些任務executor.spawn(async {println!("任務 1 開始");// 模擬異步操作futures::future::ready(()).await;println!("任務 1 完成");});executor.spawn(async {println!("任務 2 開始");// 模擬異步操作futures::future::ready(()).await;println!("任務 2 完成");});// 運行執行器executor.run();
}
事件循環與反應器（Reactor）
完整的異步運行時通常包含執行器和反應器兩部分：
- 執行器：負責調度和運行 Future
- 反應器：負責監聽 I/O 事件并喚醒相關任務
下面是一個簡化的反應器示例：
use mio::{Events, Interest, Poll, Token};
use mio::net::TcpListener;
use std::collections::HashMap;
use std::io;
use std::sync::{Arc, Mutex};
use std::task::Waker;// 簡化的反應器
struct Reactor {poll: Poll,wakers: Mutex<HashMap<Token, Waker>>,
}impl Reactor {fn new() -> io::Result<Self> {Ok(Reactor {poll: Poll::new()?,wakers: Mutex::new(HashMap::new()),})}// 注冊 I/O 資源和喚醒器fn register(&self, source: &mut TcpListener, token: Token, waker: Waker) -> io::Result<()> {self.poll.registry().register(source, token, Interest::READABLE)?;self.wakers.lock().unwrap().insert(token, waker);Ok(())}// 運行一次事件循環fn run_once(&self) -> io::Result<()> {let mut events = Events::with_capacity(1024);self.poll.poll(&mut events, None)?;for event in events.iter() {if let Some(waker) = self.wakers.lock().unwrap().get(&event.token()) {waker.wake_by_ref();}}Ok(())}
}
高級異步模式
流（Stream）處理
Stream 特征類似于 Future，但可以產生多個值：
use futures::stream::{self, StreamExt};#[tokio::main]
async fn main() {// 創建一個簡單的流let mut stream = stream::iter(1..=5);// 使用 next 方法逐個處理流中的元素while let Some(value) = stream.next().await {println!("值: {}", value);}// 使用組合子處理流let sum = stream::iter(1..=10).map(|x| x * 2).filter(|x| futures::future::ready(*x % 3 == 0)).fold(0, |acc, x| async move { acc + x }).await;println!("總和: {}", sum);
}
并發控制模式
信號量（Semaphore）
信號量用于限制并發任務數量：
use std::time::Duration;
use tokio::sync::Semaphore;
use tokio::time;#[tokio::main]
async fn main() {// 創建一個容量為 3 的信號量let semaphore = Arc::new(Semaphore::new(3));let mut handles = vec![];for id in 0..8 {let semaphore = Arc::clone(&semaphore);let handle = tokio::spawn(async move {// 獲取許可let permit = semaphore.acquire().await.unwrap();println!("任務 {} 獲取許可，開始執行", id);// 模擬工作time::sleep(Duration::from_secs(2)).await;println!("任務 {} 完成", id);// 許可在 permit 被丟棄時自動釋放drop(permit);});handles.push(handle);}// 等待所有任務完成for handle in handles {handle.await.unwrap();}
}
超時處理
為異步操作設置超時：
use std::time::Duration;
use tokio::time;async fn long_running_task() -> String {time::sleep(Duration::from_secs(5)).await;"任務完成".to_string()
}#[tokio::main]
async fn main() {// 使用 timeout 包裝異步操作match time::timeout(Duration::from_secs(2), long_running_task()).await {Ok(result) => println!("任務結果: {}", result),Err(_) => println!("任務超時"),}// 使用 select! 實現超時tokio::select! {result = long_running_task() => {println!("任務結果: {}", result);}_ = time::sleep(Duration::from_secs(2)) => {println!("任務超時");}}
}
取消和超時
在 Tokio 中實現任務取消：
use tokio::sync::oneshot;
use tokio::time;
use std::time::Duration;async fn cancelable_task(mut cancel_rx: oneshot::Receiver<()>) -> Option<String> {tokio::select! {_ = &mut cancel_rx => {println!("任務被取消");None}_ = async {// 模擬長時間運行的任務time::sleep(Duration::from_secs(3)).await;"任務完成".to_string()} => Some("任務完成".to_string()),}
}#[tokio::main]
async fn main() {// 創建取消通道let (cancel_tx, cancel_rx) = oneshot::channel();// 生成可取消的任務let handle = tokio::spawn(cancelable_task(cancel_rx));// 等待一段時間后取消任務time::sleep(Duration::from_secs(1)).await;cancel_tx.send(()).unwrap();// 等待任務完成let result = handle.await.unwrap();println!("結果: {:?}", result);
}
自定義異步 I/O
實現異步讀取器（AsyncRead）
use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::AsyncRead;// 自定義異步讀取器
struct MyAsyncReader {buffer: Vec<u8>,position: usize,
}impl MyAsyncReader {fn new(data: Vec<u8>) -> Self {MyAsyncReader {buffer: data,position: 0,}}
}impl AsyncRead for MyAsyncReader {fn poll_read(mut self: Pin<&mut Self>,_cx: &mut Context<'_>,buf: &mut tokio::io::ReadBuf<'_>,) -> Poll<io::Result<()>> {if self.position >= self.buffer.len() {return Poll::Ready(Ok(()));}let remaining = &self.buffer[self.position..];let amt = std::cmp::min(remaining.len(), buf.remaining());buf.put_slice(&remaining[..amt]);self.position += amt;Poll::Ready(Ok(()))}
}#[tokio::main]
async fn main() -> io::Result<()> {use tokio::io::AsyncReadExt;let data = b"Hello, async world!";let mut reader = MyAsyncReader::new(data.to_vec());let mut buffer = Vec::new();reader.read_to_end(&mut buffer).await?;println!("讀取的數據: {}", String::from_utf8_lossy(&buffer));Ok(())
}
實現異步寫入器（AsyncWrite）
use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::AsyncWrite;// 自定義異步寫入器
struct MyAsyncWriter {buffer: Vec<u8>,
}impl MyAsyncWriter {fn new() -> Self {MyAsyncWriter {buffer: Vec::new(),}}fn get_written_data(&self) -> &[u8] {&self.buffer}
}impl AsyncWrite for MyAsyncWriter {fn poll_write(mut self: Pin<&mut Self>,_cx: &mut Context<'_>,buf: &[u8],) -> Poll<io::Result<usize>> {self.buffer.extend_from_slice(buf);Poll::Ready(Ok(buf.len()))}fn poll_flush(self: Pin<&mut Self>,_cx: &mut Context<'_>,) -> Poll<io::Result<()>> {Poll::Ready(Ok(()))}fn poll_shutdown(self: Pin<&mut Self>,_cx: &mut Context<'_>,) -> Poll<io::Result<()>> {Poll::Ready(Ok(()))}
}#[tokio::main]
async fn main() -> io::Result<()> {use tokio::io::AsyncWriteExt;let mut writer = MyAsyncWriter::new();writer.write_all(b"Hello, async world!").await?;println!("寫入的數據: {}", String::from_utf8_lossy(writer.get_written_data()));Ok(())
}
異步設計模式
背壓（Backpressure）處理
背壓是指當生產者產生數據的速度超過消費者處理的速度時，系統需要限制生產者的速度或緩沖數據：
use tokio::sync::mpsc;
use tokio::time;
use std::time::Duration;#[tokio::main]
async fn main() {// 創建有界通道，提供背壓機制let (tx, mut rx) = mpsc::channel(5);// 生產者任務let producer = tokio::spawn(async move {for i in 1..=20 {// 當通道已滿時，send 會等待，實現背壓if let Err(e) = tx.send(i).await {println!("發送錯誤: {}", e);return;}println!("發送: {}", i);time::sleep(Duration::from_millis(100)).await;}});// 消費者任務let consumer = tokio::spawn(async move {while let Some(value) = rx.recv().await {println!("接收: {}", value);// 消費者處理較慢time::sleep(Duration::from_millis(300)).await;}});// 等待任務完成let _ = tokio::join!(producer, consumer);
}
優雅關閉（Graceful Shutdown）
實現異步系統的優雅關閉：
use tokio::signal;
use tokio::sync::{mpsc, oneshot};
use tokio::time;
use std::time::Duration;// 工作任務結構
struct Worker {id: usize,shutdown_rx: oneshot::Receiver<()>,
}impl Worker {async fn run(&mut self) {loop {tokio::select! {_ = &mut self.shutdown_rx => {println!("工作者 {} 收到關閉信號", self.id);// 執行清理操作time::sleep(Duration::from_millis(500)).await;println!("工作者 {} 已清理資源并關閉", self.id);return;}_ = time::sleep(Duration::from_secs(1)) => {println!("工作者 {} 正在處理...", self.id);}}}}
}#[tokio::main]
async fn main() {// 創建關閉通知通道let (shutdown_tx, mut shutdown_rx) = mpsc::channel(1);// 啟動工作者let mut worker_handles = vec![];let mut shutdown_senders = vec![];for i in 0..3 {let (tx, rx) = oneshot::channel();shutdown_senders.push(tx);let mut worker = Worker {id: i,shutdown_rx: rx,};worker_handles.push(tokio::spawn(async move {worker.run().await;}));}// 等待關閉信號tokio::select! {_ = signal::ctrl_c() => {println!("收到 Ctrl+C，開始優雅關閉");}_ = time::sleep(Duration::from_secs(5)) => {println!("模擬關閉信號，開始優雅關閉");}}// 發送關閉信號給所有工作者for tx in shutdown_senders {let _ = tx.send(());}// 通知關閉監聽器let _ = shutdown_tx.send(()).await;// 等待所有工作者完成for handle in worker_handles {let _ = handle.await;}println!("所有工作者已關閉，程序退出");
}
性能優化技術
任務本地存儲（Task-Local Storage）
use tokio::task::LocalKey;
use std::cell::RefCell;thread_local! {static COUNTER: RefCell<u32> = RefCell::new(0);
}#[tokio::main]
async fn main() {// 在主任務中訪問COUNTER.with(|counter| {*counter.borrow_mut() += 1;println!("主任務計數: {}", *counter.borrow());});// 在子任務中訪問let task1 = tokio::spawn(async {COUNTER.with(|counter| {*counter.borrow_mut() += 1;println!("任務 1 計數: {}", *counter.borrow());});});let task2 = tokio::spawn(async {COUNTER.with(|counter| {*counter.borrow_mut() += 1;println!("任務 2 計數: {}", *counter.borrow());});});let _ = tokio::join!(task1, task2);// 再次在主任務中訪問COUNTER.with(|counter| {println!("最終主任務計數: {}", *counter.borrow());});
}
異步棧展開（Async Stack Unwinding）
use std::panic;
use tokio::task;async fn might_panic(should_panic: bool) -> Result<(), String> {if should_panic {panic!("任務發生恐慌");}Ok(())
}#[tokio::main]
async fn main() {// 設置恐慌鉤子panic::set_hook(Box::new(|info| {println!("捕獲到恐慌: {}", info);}));// 使用 catch_unwind 捕獲異步任務中的恐慌let result = task::spawn(async {let result = panic::catch_unwind(panic::AssertUnwindSafe(|| {might_panic(true)}));match result {Ok(future) => {// 繼續處理 futureprintln!("Future 創建成功");future}Err(e) => {println!("捕獲到恐慌: {:?}", e);async { Ok(()) }}}}).await;println!("任務結果: {:?}", result);// 另一種方法：使用 tokio::spawn 的錯誤處理let handle = task::spawn(async {might_panic(true).await});match handle.await {Ok(result) => println!("任務成功: {:?}", result),Err(e) => println!("任務失敗: {}", e),}
}
實際應用案例
構建高性能 HTTP 服務器
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use std::error::Error;async fn handle_connection(mut stream: TcpStream) -> Result<(), Box<dyn Error>> {let mut buffer = [0; 1024];// 讀取請求let n = stream.read(&mut buffer).await?;let request = String::from_utf8_lossy(&buffer[..n]);println!("收到請求:\n{}", request);// 構造響應let response = "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 12\r\n\r\nHello, Rust!";// 發送響應stream.write_all(response.as_bytes()).await?;stream.flush().await?;
 觸發的處理邏輯)
![2025藍橋杯備賽Day12——P8598 [藍橋杯 2013 省 AB] 錯誤票據](http://pic.xiahunao.cn/2025藍橋杯備賽Day12——P8598 [藍橋杯 2013 省 AB] 錯誤票據)