Rust 异步编程的高级模式，涵盖并发控制、超时机制与实战架构。内容包含 Tokio 提供的异步同步原语（Mutex、RwLock、Semaphore）及其与同步版本的对比，多层超时设置与取消信号传递的实现。文章详细讲解了生产者 - 消费者、事件驱动、Actor 模型等设计模式，以及异步与同步混合编程的最佳实践。最后通过构建异步消息队列系统的实战案例，演示了依赖配置、核心实现及性能优化技巧，包括任务池、连接池使用及 Tokio 性能分析工具的应用。
💡在同步编程中，我们使用 std::sync::Mutex、std::sync::RwLock、std::sync::Semaphore 等同步原语来控制并发访问。这些原语在多线程场景下非常有效，但在异步编程中，它们会导致任务阻塞，影响性能。
异步同步原语通过 await 关键字暂停任务，而不是阻塞线程，从而提高了 CPU 利用率。Tokio 提供了一系列异步同步原语，如 tokio::sync::Mutex、tokio::sync::RwLock、tokio::sync::Semaphore。
异步 Mutex 的使用方式与标准库的类似，但需要使用 await 来获取锁。
use tokio::sync::Mutex;
use std::sync::Arc;
#[tokio::main]
async fn main() {
// 创建异步 Mutex，使用 Arc 实现共享所有权
let counter = Arc::new(Mutex::new(0));
let mut handles = vec![];
// 创建 10 个任务，每个任务增加计数器的值
for i in 0..10 {
let counter_clone = counter.clone();
let handle = tokio::spawn(async move {
// 获取锁，使用 await 暂停任务直到锁可用
let mut guard = counter_clone.lock().await;
*guard += 1;
println!("Task {}: Counter = {}", i, *guard);
// 锁会在 guard 离开作用域时自动释放
});
handles.push(handle);
}
// 等待所有任务完成
for handle in handles {
handle.await.unwrap();
}
println!("Final counter: {}", *counter.lock().await);
}
异步 RwLock 允许多个读操作同时进行，但写操作需要独占访问。
use tokio::sync::RwLock;
use std::sync::Arc;
#[tokio::main]
async fn main() {
let data = Arc::new(RwLock::new(vec![1, 2, 3]));
let mut handles = vec![];
// 创建 5 个读任务
for i in 0..5 {
let data_clone = data.clone();
let handle = tokio::spawn(async move {
let guard = data_clone.read().await;
println!("Read task {}: {:?}", i, *guard);
});
handles.push(handle);
}
// 创建 1 个写任务
let data_clone = data.clone();
let handle = tokio::spawn(async move {
let mut guard = data_clone.write().await;
guard.push(4);
println!("Write task: {:?}", *guard);
});
handles.push(handle);
// 等待所有任务完成
for handle in handles {
handle.await.unwrap();
}
println!("Final data: {:?}", *data.read().await);
}
异步 Semaphore 用于控制同时访问某一资源的任务数量。
use tokio::sync::Semaphore;
use std::sync::Arc;
#[tokio::main]
async fn main() {
// 创建信号量，允许最多 3 个任务同时访问资源
let semaphore = Arc::new(Semaphore::new(3));
let mut handles = vec![];
// 创建 10 个任务
for i in 0..10 {
let semaphore_clone = semaphore.clone();
let handle = tokio::spawn(async move {
// 获取信号量许可
let permit = semaphore_clone.acquire().await.unwrap();
println!("Task {}: Accessing resource", i);
// 模拟访问资源的耗时
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
println!("Task {}: Done", i);
// 许可会在 permit 离开作用域时自动释放
});
handles.push(handle);
}
// 等待所有任务完成
for handle in handles {
handle.await.unwrap();
}
}
| 同步原语 | 同步版本（std::sync） | 异步版本（tokio::sync） | 适用场景 |
|---|---|---|---|
| Mutex | 阻塞线程 | 暂停任务 | 保护共享数据的独占访问 |
| RwLock | 阻塞线程 | 暂停任务 | 保护共享数据的读写分离 |
| Semaphore | 阻塞线程 | 暂停任务 | 控制资源的并发访问数量 |
在复杂的异步操作中，我们可能需要设置多层超时。例如，整个操作设置一个超时，内部的子操作也设置一个更小的超时。
use tokio::time::{timeout, Duration};
async fn sub_operation() -> Result<String, String> {
tokio::time::sleep(Duration::from_secs(3)).await;
Ok("Sub operation completed".to_string())
}
async fn main_operation() -> Result<String, String> {
// 子操作设置 2 秒超时
let result = timeout(Duration::from_secs(2), sub_operation()).await;
match result {
Ok(Ok(msg)) => Ok(msg),
Ok(Err(e)) => Err(e.to_string()),
Err(_) => Err("Sub operation timeout".to_string()),
}
}
#[tokio::main]
async fn main() {
// 整个操作设置 4 秒超时
let result = timeout(Duration::from_secs(4), main_operation()).await;
match result {
Ok(Ok(msg)) => println!("Success: {}", msg),
Ok(Err(e)) => println!("Error: {}", e),
Err(_) => println!("Main operation timeout"),
}
}
当一个任务被取消时，我们可能需要通知其内部的子任务也取消，以避免资源泄漏。
use tokio::sync::oneshot;
use tokio::time::sleep;
use std::time::Duration;
async fn sub_task(mut cancel_rx: oneshot::Receiver<()>) {
println!("Sub task started");
tokio::select! {
_ = sleep(Duration::from_secs(5)) => println!("Sub task completed"),
_ = &mut cancel_rx => println!("Sub task cancelled"),
}
}
async fn main_task() {
println!("Main task started");
let (cancel_tx, cancel_rx) = oneshot::channel();
let sub_handle = tokio::spawn(sub_task(cancel_rx));
// 模拟主任务运行 3 秒后取消
sleep(Duration::from_secs(3)).await;
println!("Main task cancelling sub task");
let _ = cancel_tx.send(()); // 发送取消信号
let _ = sub_handle.await; // 等待子任务完成
println!("Main task completed");
}
#[tokio::main]
async fn main() {
main_task().await;
}
在取消任务时，我们需要确保资源被正确清理。例如，关闭文件、释放连接等。
use tokio::sync::oneshot;
use tokio::time::sleep;
use std::time::Duration;
use std::fs::File;
use std::io::Write;
async fn file_operation(mut cancel_rx: oneshot::Receiver<()>) {
println!("Opening file...");
let mut file = File::create("test.txt").unwrap();
tokio::select! {
_ = sleep(Duration::from_secs(5)) => {
file.write_all(b"Data written successfully").unwrap();
println!("File operation completed");
},
_ = &mut cancel_rx => {
println!("File operation cancelled, cleaning up...");
// 这里可以添加资源清理代码
},
}
println!("File closed");
}
#[tokio::main]
async fn main() {
let (cancel_tx, cancel_rx) = oneshot::channel();
let handle = tokio::spawn(file_operation(cancel_rx));
// 模拟 3 秒后取消
sleep(Duration::from_secs(3)).await;
let _ = cancel_tx.send(());
let _ = handle.await;
}
生产者 - 消费者模式是异步编程中最常用的模式之一。它通过一个共享的队列实现生产者和消费者之间的通信。
use tokio::sync::mpsc;
use tokio::time::sleep;
use std::time::Duration;
async fn producer(mut tx: mpsc::Sender<String>) {
for i in 0..5 {
let msg = format!("Message {}", i);
println!("Produced: {}", msg);
tx.send(msg).await.unwrap();
sleep(Duration::from_secs(1)).await;
}
// 关闭发送端
drop(tx);
}
async fn consumer(mut rx: mpsc::Receiver<String>) {
while let Some(msg) = rx.recv().await {
println!("Consumed: {}", msg);
sleep(Duration::from_secs(2)).await;
}
println!("Consumer finished");
}
#[tokio::main]
async fn main() {
// 创建通道，缓冲区大小为 2
let (tx, rx) = mpsc::channel(2);
let producer_handle = tokio::spawn(producer(tx));
let consumer_handle = tokio::spawn(consumer(rx));
producer_handle.await.unwrap();
consumer_handle.await.unwrap();
println!("Main finished");
}
事件驱动模式通过监听事件并在事件发生时触发相应的处理函数。在异步编程中，我们可以使用 tokio::stream 来实现事件驱动。
use tokio_stream::StreamExt;
use tokio::time::interval;
use std::time::Duration;
async fn event_handler(event: String) {
println!("Handling event: {}", event);
tokio::time::sleep(Duration::from_secs(1)).await;
println!("Event handled: {}", event);
}
#[tokio::main]
async fn main() {
// 创建事件流，每 2 秒产生一个事件
let event_stream = interval(Duration::from_secs(2)).map(|instant| format!("Event at {:?}", instant));
// 处理事件
event_stream.for_each(|event| async move {
event_handler(event).await;
}).await;
}
Actor 模型是一种并发模型，每个 Actor 是一个独立的执行单元，通过消息传递进行通信。在 Rust 中，我们可以使用 tokio::sync::oneshot 或 tokio::sync::mpsc 实现简单的 Actor。
use tokio::sync::{mpsc, oneshot};
use std::collections::HashMap;
// Actor 的消息类型
enum ActorMessage {
Insert { key: String, value: String, reply: oneshot::Sender<()> },
Get { key: String, reply: oneshot::Sender<Option<String>> },
Remove { key: String, reply: oneshot::Sender<()> },
}
async fn actor(mut rx: mpsc::Receiver<ActorMessage>) {
let mut store = HashMap::new();
while let Some(msg) = rx.recv().await {
match msg {
ActorMessage::Insert { key, value, reply } => {
store.insert(key, value);
let _ = reply.send(());
},
ActorMessage::Get { key, reply } => {
let value = store.get(&key).cloned();
let _ = reply.send(value);
},
ActorMessage::Remove { key, reply } => {
store.remove(&key);
let _ = reply.send(());
},
}
}
}
struct ActorClient {
tx: mpsc::Sender<ActorMessage>,
}
impl ActorClient {
pub fn new(tx: mpsc::Sender<ActorMessage>) -> Self {
ActorClient { tx }
}
pub async fn insert(&self, key: String, value: String) {
let (reply_tx, reply_rx) = oneshot::channel();
self.tx.send(ActorMessage::Insert { key, value, reply: reply_tx }).await.unwrap();
reply_rx.await.unwrap();
}
pub async fn get(&self, key: String) -> Option<String> {
let (reply_tx, reply_rx) = oneshot::channel();
self.tx.send(ActorMessage::Get { key, reply: reply_tx }).await.unwrap();
reply_rx.await.unwrap()
}
pub async fn remove(&self, key: String) {
let (reply_tx, reply_rx) = oneshot::channel();
self.tx.send(ActorMessage::Remove { key, reply: reply_tx }).await.unwrap();
reply_rx.await.unwrap();
}
}
#[tokio::main]
async fn main() {
let (tx, rx) = mpsc::channel(32);
let _actor_handle = tokio::spawn(actor(rx));
let client = ActorClient::new(tx);
client.insert("key1".to_string(), "value1".to_string()).await;
println!("Get key1: {:?}", client.get("key1".to_string()).await);
client.insert("key2".to_string(), "value2".to_string()).await;
println!("Get key2: {:?}", client.get("key2".to_string()).await);
client.remove("key1".to_string()).await;
println!("Get key1 after remove: {:?}", client.get("key1".to_string()).await);
}
在异步代码中调用同步代码可能会导致任务阻塞，影响性能。Tokio 提供了 spawn_blocking 函数，可以将同步代码放在单独的线程池中执行，避免阻塞异步任务。
use tokio::task::spawn_blocking;
fn sync_operation() -> String {
// 模拟耗时的同步操作
std::thread::sleep(std::time::Duration::from_secs(2));
"Sync operation completed".to_string()
}
async fn async_operation() -> String {
// 使用 spawn_blocking 执行同步代码
let result = spawn_blocking(sync_operation).await.unwrap();
println!("Sync operation result: {}", result);
"Async operation completed".to_string()
}
#[tokio::main]
async fn main() {
println!("Start");
let result = async_operation().await;
println!("Result: {}", result);
println!("End");
}
在同步代码中调用异步代码需要使用 block_on 函数，它会阻塞当前线程直到异步操作完成。
use tokio::runtime::Runtime;
async fn async_operation() -> String {
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
"Async operation completed".to_string()
}
fn sync_main() {
println!("Start");
// 创建一个 Tokio 运行时
let rt = Runtime::new().unwrap();
// 使用 block_on 阻塞当前线程直到异步操作完成
let result = rt.block_on(async_operation());
println!("Result: {}", result);
println!("End");
}
fn main() {
sync_main();
}
我们将构建一个简单的异步消息队列系统，支持以下功能：
项目架构设计：
创建项目：
cargo new rust-async-queue
cd rust-async-queue
在 Cargo.toml 中添加依赖：
[dependencies]
tokio = { version = "1.0", features = ["full"] }
redis = { version = "0.22", features = ["tokio-comp"] }
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
创建 src/models.rs：
use serde::{Deserialize, Serialize};
#[derive(Debug, Serialize, Deserialize, Clone)]
pub struct Message {
pub id: String,
pub content: String,
pub retry_count: u32,
pub max_retries: u32,
pub created_at: chrono::DateTime<chrono::Utc>,
}
impl Message {
pub fn new(content: String, max_retries: u32) -> Self {
Message {
id: uuid::Uuid::new_v4().to_string(),
content,
retry_count: 0,
max_retries,
created_at: chrono::Utc::now(),
}
}
pub fn should_retry(&self) -> bool {
self.retry_count < self.max_retries
}
pub fn increment_retry_count(&mut self) {
self.retry_count += 1;
}
}
创建 src/queue.rs：
use redis::AsyncCommands;
use serde_json;
use crate::models::Message;
pub struct AsyncQueue {
client: redis::Client,
queue_name: String,
dead_letter_queue: String,
}
impl AsyncQueue {
pub fn new(url: &str, queue_name: &str) -> Self {
AsyncQueue {
client: redis::Client::open(url).unwrap(),
queue_name: queue_name.to_string(),
dead_letter_queue: format!("{}:dead", queue_name),
}
}
pub async fn enqueue(&self, message: Message) -> Result<(), String> {
let mut conn = self.client.get_async_connection().await.map_err(|e| e.to_string())?;
let message_json = serde_json::to_string(&message).map_err(|e| e.to_string())?;
conn.rpush(&self.queue_name, message_json).await.map_err(|e| e.to_string())?;
Ok(())
}
pub async fn dequeue(&self) -> Result<Option<Message>, String> {
let mut conn = self.client.get_async_connection().await.map_err(|e| e.to_string())?;
let result: Option<String> = conn.blpop(&self.queue_name, 5).await.map_err(|e| e.to_string())?.map(|(_, v)| v);
match result {
Some(message_json) => {
let message = serde_json::from_str(&message_json).map_err(|e| e.to_string())?;
Ok(Some(message))
},
None => Ok(None),
}
}
pub async fn enqueue_dead_letter(&self, message: Message) -> Result<(), String> {
let mut conn = self.client.get_async_connection().await.map_err(|e| e.to_string())?;
let message_json = serde_json::to_string(&message).map_err(|e| e.to_string())?;
conn.rpush(&self.dead_letter_queue, message_json).await.map_err(|e| e.to_string())?;
Ok(())
}
pub async fn len(&self) -> Result<usize, String> {
let mut conn = self.client.get_async_connection().await.map_err(|e| e.to_string())?;
let len: usize = conn.llen(&self.queue_name).await.map_err(|e| e.to_string())?;
Ok(len)
}
}
创建 src/producer.rs：
use crate::models::Message;
use crate::queue::AsyncQueue;
pub struct Producer {
queue: AsyncQueue,
}
impl Producer {
pub fn new(queue: AsyncQueue) -> Self {
Producer { queue }
}
pub async fn send(&self, content: String, max_retries: u32) -> Result<(), String> {
let message = Message::new(content, max_retries);
self.queue.enqueue(message).await
}
}
创建 src/consumer.rs：
use crate::models::Message;
use crate::queue::AsyncQueue;
pub struct Consumer {
queue: AsyncQueue,
handler: Box<dyn Fn(Message) -> Box<dyn std::future::Future<Output = Result<(), String>> + Send>>,
}
impl Consumer {
pub fn new(
queue: AsyncQueue,
handler: Box<dyn Fn(Message) -> Box<dyn std::future::Future<Output = Result<(), String>> + Send>>,
) -> Self {
Consumer { queue, handler }
}
pub async fn start(mut self) {
println!("Consumer started");
loop {
match self.queue.dequeue().await {
Ok(Some(mut message)) => {
println!("Consumed message: {}", message.content);
let result = (self.handler)(message.clone()).await;
match result {
Ok(()) => println!("Message processed successfully: {}", message.content),
Err(e) => {
println!("Error processing message: {} - {}", message.content, e);
if message.should_retry() {
message.increment_retry_count();
println!("Retrying message ({} of {}): {}", message.retry_count, message.max_retries, message.content);
self.queue.enqueue(message).await.unwrap();
} else {
println!("Message failed after {} retries: {}", message.max_retries, message.content);
self.queue.enqueue_dead_letter(message).await.unwrap();
}
},
}
},
Ok(None) => continue,
Err(e) => {
println!("Error dequeuing message: {}", e);
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
},
}
}
}
}
创建 src/main.rs：
use tokio;
use crate::consumer::Consumer;
use crate::producer::Producer;
use crate::queue::AsyncQueue;
mod consumer;
mod models;
mod producer;
mod queue;
#[tokio::main]
async fn main() {
let queue = AsyncQueue::new("redis://127.0.0.1/", "test_queue");
// 创建生产者
let producer = Producer::new(queue.clone());
// 创建消费者
let consumer = Consumer::new(queue.clone(), Box::new(|message| {
Box::new(async move {
println!("Processing message: {}", message.content);
// 模拟处理消息的耗时
tokio::time::sleep(std::time::Duration::from_secs(2)).await;
// 随机返回成功或失败
if rand::random() {
Ok(())
} else {
Err("Processing failed".to_string())
}
})
}));
// 启动消费者
tokio::spawn(consumer.start());
// 生产者发送消息
for i in 0..10 {
let content = format!("Message {}", i);
producer.send(content, 3).await.unwrap();
println!("Sent message {}", i);
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
}
// 等待程序结束
tokio::time::sleep(std::time::Duration::from_secs(30)).await;
}
我们可以使用 Redis 的 INFO 命令查看队列的性能指标。例如，查看 keyspace_hits 和 keyspace_misses：
redis-cli INFO keyspace
或者使用第三方工具，如 Redis-benchmark，测试队列的读写性能：
redis-benchmark -h 127.0.0.1 -p 6379 -c 100 -n 10000 -t rpush,blpop
在异步编程中，频繁创建任务会导致大量的内存分配和上下文切换。我们可以通过以下方式减少任务创建的开销：
use tokio::sync::mpsc;
use tokio::time::sleep;
use std::time::Duration;
async fn process_batch(messages: Vec<String>) {
println!("Processing batch of {} messages", messages.len());
// 模拟处理消息的耗时
sleep(Duration::from_secs(1)).await;
for msg in messages {
println!("Processed: {}", msg);
}
}
#[tokio::main]
async fn main() {
let (tx, mut rx) = mpsc::channel(100);
// 生产者发送消息
tokio::spawn(async move {
for i in 0..10 {
tx.send(format!("Message {}", i)).await.unwrap();
sleep(Duration::from_millis(500)).await;
}
drop(tx);
});
// 消费者批量处理消息
let mut batch = Vec::new();
while let Some(msg) = rx.recv().await {
batch.push(msg);
if batch.len() >= 5 {
process_batch(batch).await;
batch = Vec::new();
}
}
if !batch.is_empty() {
process_batch(batch).await;
}
println!("All messages processed");
}
对于数据库、Redis 等资源，使用连接池可以避免频繁创建和销毁连接。Tokio 提供了 tokio::sync::Semaphore 或第三方库（如 deadpool）来实现连接池。
use tokio::sync::Semaphore;
use std::sync::Arc;
use sqlx::PgPool;
async fn create_pool(database_url: &str) -> PgPool {
let max_connections = 10;
PgPool::connect(database_url).await.unwrap()
}
async fn query_data(pool: Arc<PgPool>) {
let mut conn = pool.acquire().await.unwrap();
let result = sqlx::query!("SELECT * FROM users").fetch_all(&mut conn).await.unwrap();
println!("Query result: {:?}", result);
}
#[tokio::main]
async fn main() {
let pool = Arc::new(create_pool("postgresql://user:password@localhost:5432/mydb").await);
let mut handles = vec![];
for i in 0..20 {
let pool_clone = pool.clone();
let handle = tokio::spawn(async move {
query_data(pool_clone).await;
});
handles.push(handle);
}
for handle in handles {
handle.await.unwrap();
}
}
在异步代码中，不必要的 await 会导致任务调度的开销。我们应该尽量避免在不需要等待的地方使用 await。
use tokio::time::sleep;
use std::time::Duration;
async fn task1() -> i32 {
sleep(Duration::from_secs(1)).await;
1
}
async fn task2() -> i32 {
sleep(Duration::from_secs(2)).await;
2
}
async fn process() -> i32 {
let result1 = task1().await;
let result2 = task2().await;
result1 + result2
}
#[tokio::main]
async fn main() {
let start = std::time::Instant::now();
let result = process().await;
println!("Result: {}, Time taken: {:?}", result, start.elapsed());
}
优化后的代码：
use tokio::join;
use tokio::time::sleep;
use std::time::Duration;
async fn task1() -> i32 {
sleep(Duration::from_secs(1)).await;
1
}
async fn task2() -> i32 {
sleep(Duration::from_secs(2)).await;
2
}
async fn process() -> i32 {
let (result1, result2) = join!(task1(), task2());
result1 + result2
}
#[tokio::main]
async fn main() {
let start = std::time::Instant::now();
let result = process().await;
println!("Result: {}, Time taken: {:?}", result, start.elapsed());
}
Tokio 提供了一些性能分析工具，如 tokio-console 和 tracing，可以帮助我们定位和优化性能问题。
使用 tokio-console 可以可视化异步任务的调度和执行情况：
cargo install tokio-console
RUSTFLAGS="--cfg tokio_unstable" cargo run # 在另一个终端运行 tokio-console
使用 tracing 可以记录异步代码的执行情况，帮助我们定位性能瓶颈：
[dependencies]
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter", "json"] }
use tracing::info;
use tracing_subscriber::prelude::*;
#[tokio::main]
async fn main() {
tracing_subscriber::registry()
.with(tracing_subscriber::EnvFilter::new("info"))
.with(tracing_subscriber::fmt::layer())
.init();
info!("Start program");
let start = std::time::Instant::now();
process().await;
info!("Program completed in {:?}", start.elapsed());
}
async fn process() {
info!("Processing...");
task1().await;
task2().await;
}
Rust 的异步编程提供了高性能、内存安全的并发处理能力。通过深入理解异步同步原语、超时与取消的高级用法、异步编程设计模式、异步与同步的混合编程以及性能优化方法，我们可以编写出更高效、更安全的异步代码。
在实际项目中，我们需要注意异步编程的常见错误，如任务泄漏、死锁、非 Send 类型的使用等，并遵循最佳实践，如使用异步同步原语、正确处理取消、使用 spawn_blocking 处理同步操作等。同时，我们可以使用 Tokio 的性能分析工具来定位和优化性能问题。
希望本章的内容能够帮助您深入掌握 Rust 异步编程的核心技术，并在实际项目中应用。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 zeeklog
将字符串编码和解码为其 Base64 格式表示形式即可。 在线工具，Base64 字符串编码/解码在线工具，online
将字符串、文件或图像转换为其 Base64 表示形式。 在线工具，Base64 文件转换器在线工具，online
将 Markdown（GFM）转为 HTML 片段，浏览器内 marked 解析；与 HTML 转 Markdown 互为补充。 在线工具，Markdown 转 HTML在线工具，online
将 HTML 片段转为 GitHub Flavored Markdown，支持标题、列表、链接、代码块与表格等；浏览器内处理，可链接预填。 在线工具，HTML 转 Markdown在线工具，online
通过删除不必要的空白来缩小和压缩JSON。 在线工具，JSON 压缩在线工具，online
将JSON字符串修饰为友好的可读格式。 在线工具，JSON美化和格式化在线工具，online