rust

Rust 异步与并发编程：Tokio、async/await 与并发模式

Rust 异步编程全貌：async/await 语法糖背后的 Poll 模型、Tokio 运行时架构、Channel 通信模式、并发安全 trait（Send/Sync）、流处理（Stream/Sink）与背压控制、真实高并发服务器实现。

Leeting Yan 2025-08-01 4 分钟阅读 1835 字

1. Rust 并发模型概览

Rust 提供三种并发方式：

方式	API	特点
OS 线程	`std::thread`	真并行，开销大
异步任务	`tokio::spawn`	协程级轻量，I/O 密集
并行计算	`rayon::join`	数据并行，CPU 密集

// OS 线程
use std::thread;
let handle = thread::spawn(|| { /* 工作 */ });
handle.join().unwrap();

2. async/await 与 Poll 模型

2.1 async 函数的本质

async fn 是语法糖，编译后生成一个实现 Future trait 的状态机：

async fn fetch_user(id: u32) -> User {
    let db = connect_db().await;        // 状态点 1
    let user = db.query(id).await;      // 状态点 2
    user
}

// 等价于（编译器生成）
enum FetchUserFuture {
    Start(u32),
    Connecting(u32, ConnectDbFuture),
    Querying(u32, DbConnection, QueryFuture),
    Done,
}

2.2 Future trait 原理

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() 非阻塞，返回 Ready 或 Pending
Pending 时注册 waker，事件发生后被重新调度

2.3 Pin 与自引用结构

async 状态机可能自引用，需要 Pin 保证内存地址不变：

use std::pin::Pin;

let mut future = fetch_user(1);
let pinned = Pin::new(&mut future);
// pinned.poll(...); // Pin 保证状态机内部指针有效

2.4 阻塞操作会阻塞整个线程

// ❌ 错误：在 async 中阻塞
async fn bad() {
    std::thread::sleep(Duration::from_secs(1)); // 阻塞当前线程！
}

// ✅ 正确：使用 tokio 的非阻塞 API
async fn good() {
    tokio::time::sleep(Duration::from_secs(1)).await; // 让出线程
}

3. Tokio 运行时架构

3.1 多线程运行时（默认）

#[tokio::main]
async fn main() {
    // 等效于：
    // tokio::runtime::Runtime::new().unwrap().block_on(async { ... })
}

Tokio 运行时组成：

任务调度器：work-stealing 队列，多线程调度 async 任务
I/O 驱动：epoll/kqueue/IOCP 事件循环
定时器：高效的时间轮实现
阻塞线程池：spawn_blocking 执行阻塞操作

3.2 核心 API

use tokio::task;

// 创建任务
let handle = tokio::spawn(async { 42 });
let result = handle.await.unwrap();

// 并发执行多个 Future
let (a, b) = tokio::join!(
    fetch_user(1),
    fetch_user(2),
);

// 竞争：先完成的胜出
let result = tokio::select! {
    user = fetch_user(1) => user,
    post = fetch_post(1) => post,
};

// 带超时
let result = tokio::time::timeout(
    Duration::from_secs(5),
    fetch_user(1)
).await;

3.3 spawn_blocking

将阻塞操作放到独立线程池：

let result = task::spawn_blocking(|| {
    std::fs::read_to_string("file.txt") // 阻塞文件 I/O
}).await.unwrap();

4. Channel 通信模式

Tokio 提供多种 channel 类型：

类型	容量	用途
`mpsc`	有界/无界	多生产者单消费者
`oneshot`	1	一次性响应
`broadcast`	有界	多播，所有接收者收到副本
`watch`	1	状态广播，只保留最新值

4.1 mpsc Channel

use tokio::sync::mpsc;

let (tx, mut rx) = mpsc::channel(100); // 缓冲容量 100

// 发送端 clone 给多个生产者
let tx2 = tx.clone();

tokio::spawn(async move {
    tx.send("msg1").await.unwrap();
});

tokio::spawn(async move {
    tx2.send("msg2").await.unwrap();
});

while let Some(msg) = rx.recv().await {
    println!("received: {}", msg);
}

4.2 broadcast Channel

use tokio::sync::broadcast;

let (tx, _rx) = broadcast::channel(16);
let mut rx1 = tx.subscribe();
let mut rx2 = tx.subscribe();

tokio::spawn(async move {
    println!("rx1: {:?}", rx1.recv().await);
});
tokio::spawn(async move {
    println!("rx2: {:?}", rx2.recv().await);
});

tx.send("hello all").unwrap();

4.3 oneshot：请求-响应模式

use tokio::sync::oneshot;

let (tx, rx) = oneshot::channel();

tokio::spawn(async move {
    let result = compute().await;
    tx.send(result).unwrap();
});

let result = rx.await.unwrap();

4.4 watch：配置热更新

use tokio::sync::watch;

let (tx, rx) = watch::channel("config v1");

// 订阅者
let mut rx_clone = rx.clone();
tokio::spawn(async move {
    while rx_clone.changed().await.is_ok() {
        println!("config updated: {}", *rx_clone.borrow());
    }
});

// 更新配置
tx.send("config v2").unwrap();

5. 并发安全：Send 与 Sync

5.1 自动推导规则

Send：可以安全地将所有权转移到另一个线程
Sync：可以安全地在多个线程间共享引用（即 &T 是 Send）

几乎所有类型都自动实现这两个 trait，编译器会自动推导。例外情况：

Rc<T>：非线程安全引用计数 → 非 Send + 非 Sync
RefCell<T>：运行时借用检查无锁保护 → 非 Sync
*const T / *mut T：裸指针 → 非 Send + 非 Sync

5.2 跨线程传递数据

use std::sync::Arc;
use tokio::sync::Mutex;

// ❌ Rc<RefCell<i32>> 不能跨线程
// let data = Rc::new(RefCell::new(0));

// ✅ Arc<Mutex<i32>> 可以跨线程
let data = Arc::new(Mutex::new(0));

for _ in 0..10 {
    let d = Arc::clone(&data);
    tokio::spawn(async move {
        let mut guard = d.lock().await;
        *guard += 1;
    });
}

6. Stream 与背压控制

6.1 Stream trait

Stream 是异步版本的 Iterator：

use tokio_stream::StreamExt;

let mut stream = tokio_stream::iter(vec![1, 2, 3, 4, 5]);

while let Some(item) = stream.next().await {
    println!("{}", item);
}

6.2 背压控制（Backpressure）

use tokio::sync::mpsc;
use tokio_stream::wrappers::ReceiverStream;

let (tx, rx) = mpsc::channel(10); // 有界 channel = 背压

// 生产者不会因为消费慢而无限堆积内存
tokio::spawn(async move {
    for i in 0..1000 {
        tx.send(i).await.unwrap(); // 满时自动等待
    }
});

let mut stream = ReceiverStream::new(rx);
while let Some(item) = stream.next().await {
    tokio::time::sleep(Duration::from_millis(10)).await; // 慢消费
}

6.3 限制并发度

use futures::stream::{self, StreamExt};

stream::iter(urls)
    .map(|url| reqwest::get(url))        // 创建 Future
    .buffer_unordered(10)                 // 最多 10 个并发请求
    .for_each(|response| async {
        println!("{}", response.unwrap().status());
    })
    .await;

7. 并发模式实战

7.1 Actor 模式

struct Actor {
    receiver: mpsc::Receiver<Message>,
    state: HashMap<u64, User>,
}

enum Message {
    GetUser(u64, oneshot::Sender<Option<User>>),
    SetUser(u64, User),
}

impl Actor {
    async fn run(mut self) {
        while let Some(msg) = self.receiver.recv().await {
            match msg {
                Message::GetUser(id, tx) => {
                    tx.send(self.state.get(&id).cloned()).unwrap();
                }
                Message::SetUser(id, user) => {
                    self.state.insert(id, user);
                }
            }
        }
    }
}

7.2 扇出-扇入（Fan-out / Fan-in）

// 扇出：一个生产者 → 多个消费者
let (tx, _rx) = broadcast::channel(100);
for i in 0..4 {
    let mut rx = tx.subscribe();
    tokio::spawn(async move {
        while let Ok(msg) = rx.recv().await {
            process(msg).await;
        }
    });
}

// 扇入：多个生产者 → 一个消费者
let (tx, mut rx) = mpsc::channel(100);
for i in 0..4 {
    let tx = tx.clone();
    tokio::spawn(async move { tx.send(i).await; });
}
drop(tx); // 关闭原始 tx，当所有 clone drop 后 rx 结束

7.3 优雅关闭（Graceful Shutdown）

use tokio::signal;

let (shutdown_tx, mut shutdown_rx) = mpsc::channel(1);

let server = tokio::spawn(async move {
    tokio::select! {
        _ = run_server() => {},
        _ = shutdown_rx.recv() => {
            println!("Shutting down gracefully...");
            cleanup().await;
        }
    }
});

// 捕获 Ctrl+C
signal::ctrl_c().await.unwrap();
shutdown_tx.send(()).await.unwrap();
server.await.unwrap();

8. 性能调优与陷阱

8.1 避免在 async 中持有锁过长时间

// ❌ 持有锁跨越 await 点
async fn bad(cache: &Mutex<HashMap<u64, User>>) {
    let mut guard = cache.lock().await;
    let user = guard.get(&1).cloned();
    let data = fetch_remote().await; // 锁期间阻塞其他任务！
    guard.insert(1, data);
}

// ✅ 缩小锁的作用域
async fn good(cache: &Mutex<HashMap<u64, User>>) {
    let maybe_user = {
        let guard = cache.lock().await;
        guard.get(&1).cloned()
    };
    let data = fetch_remote().await;
    cache.lock().await.insert(1, data);
}

8.2 使用 tokio::task::yield_now 协作调度

async fn cpu_heavy() {
    for i in 0..1_000_000 {
        compute(i);
        if i % 10_000 == 0 {
            tokio::task::yield_now().await; // 让出执行权
        }
    }
}

8.3 线程池大小配置

use tokio::runtime;

let rt = runtime::Builder::new_multi_thread()
    .worker_threads(8)      // 根据 CPU 核心数调整
    .max_blocking_threads(512)
    .enable_all()
    .build()
    .unwrap();

8.4 常见陷阱

陷阱	症状	解决方案
在 async 中使用阻塞 I/O	吞吐量骤降	换 tokio 异步 API 或 spawn_blocking
`Mutex` 跨越 await	死锁或低并发	缩小锁作用域，或用 `tokio::sync::RwLock`
无界 channel 堆积	内存 OOM	使用有界 channel
忘记 await Future	任务未执行	`let _ = future.await;` 或显式 spawn

Tokio + async/await 使 Rust 在 I/O 密集场景下达到与 Go、Node.js 相当的开发效率，同时保持零成本抽象和编译期安全保证。掌握这些模式是构建高性能网络服务的关键。

← 上一篇 Rust 所有权与内存安全：智能指针、内部可变性与 Unsafe 边界 下一篇 → Rust Web 框架实战：Axum、Actix-web 与 Rocket 深度对比

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页