Rust 异步编程的核心概念与实践。内容涵盖 Future、Poll、Waker 机制，async/await 语法基础，以及 Tokio 运行时架构。通过 TCP/UDP、文件 IO、Redis 及 HTTP 请求等示例演示异步 IO 操作。此外，文章还展示了基于 Axum 和 SQLx 构建高性能 RESTful API 服务的完整流程，包括依赖配置、路由处理、Docker 部署及性能测试。最后总结了异步编程的常见错误、优化技巧与生态推荐，旨在帮助开发者掌握 Rust 异步技术并应用于实际项目。
💡在传统的同步编程中，代码按照顺序执行，每个操作必须等待前一个完成才能继续。例如，发送网络请求时，主线程会阻塞直到响应返回，这种方式简单直观，但在高并发场景下效率低下，因为大量线程会因阻塞而闲置。
异步编程则允许代码在等待操作完成时继续执行其他任务。当一个异步操作开始后，程序会立即返回并继续处理下一个任务，直到该操作完成后通过回调或事件通知继续执行后续代码。这种方式显著提高了 CPU 利用率和系统的并发处理能力。
Rust 的异步编程经历了几个重要阶段：
futures 库提供基础的 Future 和 Executor 支持，但语法冗长且难以使用。async/await 语法糖的实验版本，简化了异步代码的编写。async/await 正式稳定，成为 Rust 异步编程的标准范式。在 Rust 异步编程中，有三个核心概念：
Future 是异步操作的抽象表示，它代表一个尚未完成但最终会产生结果的计算。在 Rust 中，Future 是一个 trait，定义如下：
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
pub trait Future {
type Output;
fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output>;
}
💡Output 是异步操作的结果类型。poll 方法是 Future 的核心，它尝试推进异步操作的执行。
Poll 是 poll 方法的返回类型，用于表示异步操作的状态：
Poll::Ready(value)：异步操作已完成，返回结果 value。Poll::Pending：异步操作尚未完成，需要再次调用 poll 方法。当 Future 返回 Pending 时，它需要一种方式通知 Executor（执行器）在操作完成后再次轮询。Waker 就是负责这个通知机制的组件。在 Context 结构体中可以获取到 Waker 实例，Future 可以在需要时保存 Waker，以便操作完成后唤醒 Executor。
async 关键字用于将代码块或函数标记为异步。异步函数的返回值是一个实现了 Future trait 的匿名类型。例如：
async fn fetch_data() -> String {
// 模拟网络请求
"Hello, async!".to_string()
}
fn main() {
let future = fetch_data(); // 这里并没有执行异步代码
println!("Future created");
}
💡调用异步函数只会创建一个 Future 对象，不会立即执行代码。我们需要一个 Executor 来驱动 Future 的执行。
await 关键字用于暂停异步代码的执行，直到 Future 完成。await 必须在 async 函数或 async 代码块中使用。例如：
async fn fetch_data() -> String {
"Hello, async!".to_string()
}
async fn process() {
println!("Start processing");
let data = fetch_data().await; // 暂停执行，等待 fetch_data 完成
println!("Data received: {}", data);
}
#[tokio::main]
async fn main() {
process().await; // 驱动 process() 的执行
}
⚠️注意：await 关键字会自动处理 Future 的状态和唤醒机制，让异步代码看起来像同步代码一样。
异步函数的返回类型是一个实现了 Future trait 的匿名类型。如果需要显式指定返回类型，可以使用 Box<dyn Future<Output = T>>。例如：
use std::future::Future;
fn fetch_data() -> impl Future<Output = String> {
async move {
"Hello, async!".to_string()
}
}
#[tokio::main]
async fn main() {
let data = fetch_data().await;
println!("{}", data);
}
Rust 的异步代码需要一个 Executor 来执行。Executor 负责管理任务的调度、唤醒和执行。常用的 Executor 有 Tokio 和 async-std。例如，使用 Tokio 的 #[tokio::main] 宏可以轻松启动一个异步执行器：
use tokio;
#[tokio::main]
async fn main() {
println!("Hello from async main"); // 异步代码
}
Tokio 是 Rust 最流行的异步运行时库，提供了高性能的网络 IO、定时器、任务调度等功能。它的架构主要由以下组件构成：
在 Cargo.toml 中添加 Tokio 依赖：
[dependencies]
tokio = { version = "1.0", features = ["full"] }
💡features = ["full"] 会启用 Tokio 的所有功能。如果需要更小的二进制文件，可以只启用所需的功能，如 ["rt-multi-thread", "macros"]。
use tokio::net::TcpListener;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
#[tokio::main]
async fn main() {
let listener = TcpListener::bind("127.0.0.1:8080").await.unwrap();
println!("Server listening on 127.0.0.1:8080");
loop {
let (mut socket, addr) = listener.accept().await.unwrap();
println!("Accepted connection from {}", addr);
tokio::spawn(async move {
let mut buf = [0; 1024];
loop {
match socket.read(&mut buf).await {
Ok(0) => {
println!("Client {} disconnected", addr);
return;
}
Ok(n) => {
let data = &buf[..n];
println!("Received from {}: {}", addr, String::from_utf8_lossy(data));
socket.write_all(data).await.unwrap();
}
Err(e) => {
println!("Error reading from client {}: {}", addr, e);
return;
}
}
}
});
}
}
use tokio::net::TcpStream;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
#[tokio::main]
async fn main() {
let mut stream = TcpStream::connect("127.0.0.1:8080").await.unwrap();
println!("Connected to server");
let message = "Hello, Tokio!";
stream.write_all(message.as_bytes()).await.unwrap();
println!("Sent: {}", message);
let mut buf = [0; 1024];
let n = stream.read(&mut buf).await.unwrap();
let response = String::from_utf8_lossy(&buf[..n]);
println!("Received: {}", response);
}
Tokio 提供了任务调度功能，可以创建轻量级的异步任务。任务分为两种类型：
#[tokio::main] 宏创建的主任务。tokio::spawn 函数创建的任务。任务调度器会自动管理任务的执行，确保 CPU 资源的高效利用。例如：
use tokio;
use std::time::Duration;
#[tokio::main]
async fn main() {
// 创建三个子任务
let task1 = tokio::spawn(async {
tokio::time::sleep(Duration::from_secs(1)).await;
println!("Task 1 completed");
});
let task2 = tokio::spawn(async {
tokio::time::sleep(Duration::from_secs(2)).await;
println!("Task 2 completed");
});
let task3 = tokio::spawn(async {
tokio::time::sleep(Duration::from_secs(3)).await;
println!("Task 3 completed");
});
// 等待所有任务完成
task1.await.unwrap();
task2.await.unwrap();
task3.await.unwrap();
println!("All tasks completed");
}
Tokio 提供了异步文件 IO 的支持。例如，读取文件内容：
use tokio::fs::File;
use tokio::io::AsyncReadExt;
#[tokio::main]
async fn main() {
let mut file = File::open("test.txt").await.unwrap();
let mut content = String::new();
file.read_to_string(&mut content).await.unwrap();
println!("File content: {}", content);
}
除了 TCP，Tokio 还支持 UDP、Unix 域套接字等网络协议。例如，UDP 服务器：
use tokio::net::UdpSocket;
#[tokio::main]
async fn main() {
let socket = UdpSocket::bind("127.0.0.1:8080").await.unwrap();
println!("UDP server listening on 127.0.0.1:8080");
let mut buf = [0; 1024];
loop {
let (n, addr) = socket.recv_from(&mut buf).await.unwrap();
println!("Received from {}: {}", addr, String::from_utf8_lossy(&buf[..n]));
socket.send_to(&buf[..n], addr).await.unwrap();
}
}
redis-rs 库提供了异步 Redis 操作的支持。在 Cargo.toml 中添加依赖：
[dependencies]
redis = { version = "0.22", features = ["tokio-comp"] }
示例代码：
use redis::AsyncCommands;
#[tokio::main]
async fn main() {
let client = redis::Client::open("redis://127.0.0.1/").unwrap();
let mut conn = client.get_async_connection().await.unwrap();
// 设置键值对
let _: () = conn.set("key", "value").await.unwrap();
println!("Set key:value");
// 获取键值对
let value: String = conn.get("key").await.unwrap();
println!("Get key: {}", value);
// 删除键值对
let _: () = conn.del("key").await.unwrap();
println!("Del key");
}
reqwest 是 Rust 常用的 HTTP 客户端库，支持异步操作。在 Cargo.toml 中添加依赖：
[dependencies]
reqwest = "0.11"
tokio = { version = "1.0", features = ["full"] }
示例代码：
use reqwest;
#[tokio::main]
async fn main() {
let url = "https://httpbin.org/get";
let response = reqwest::get(url).await.unwrap();
println!("Status: {}", response.status());
let body = response.text().await.unwrap();
println!("Body: {}", body);
}
在异步编程中，Future 可能包含自引用结构体，即结构体的某个字段指向自身的另一个字段。这种情况下，Future 的内存地址必须固定，否则当 Future 被移动时，引用会失效。Pin trait 就是用来确保 Future 不会被移动的。
Unpin trait 表示 Future 可以安全地移动。大多数类型默认实现了 Unpin，只有包含自引用的类型需要显式实现 Pin。例如：
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
struct MyFuture {
done: bool,
data: Option<String>,
}
impl Future for MyFuture {
type Output = String;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
if self.done {
Poll::Ready(self.data.take().unwrap())
} else {
self.done = true;
self.data = Some("Data from MyFuture".to_string());
cx.waker().wake_by_ref(); // 唤醒自己
Poll::Pending
}
}
}
#[tokio::main]
async fn main() {
let future = MyFuture { done: false, data: None };
let result = future.await;
println!("{}", result);
}
Future 的组合操作可以让我们并发执行多个异步任务，并对结果进行处理。Tokio 提供了多种组合器：
use tokio::join;
use std::time::Duration;
async fn task1() -> String {
tokio::time::sleep(Duration::from_secs(1)).await;
"Task 1 completed".to_string()
}
async fn task2() -> String {
tokio::time::sleep(Duration::from_secs(2)).await;
"Task 2 completed".to_string()
}
#[tokio::main]
async fn main() {
let (res1, res2) = join!(task1(), task2());
println!("{}", res1);
println!("{}", res2);
}
use tokio::select;
use std::time::Duration;
async fn task1() -> String {
tokio::time::sleep(Duration::from_secs(1)).await;
"Task 1 completed".to_string()
}
async fn task2() -> String {
tokio::time::sleep(Duration::from_secs(2)).await;
"Task 2 completed".to_string()
}
#[tokio::main]
async fn main() {
select! {
res1 = task1() => println!("{}", res1),
res2 = task2() => println!("{}", res2),
}
}
use tokio::try_join;
use std::time::Duration;
async fn task1() -> Result<String, String> {
tokio::time::sleep(Duration::from_secs(1)).await;
Ok("Task 1 completed".to_string())
}
async fn task2() -> Result<String, String> {
tokio::time::sleep(Duration::from_secs(2)).await;
Err("Task 2 failed".to_string())
}
#[tokio::main]
async fn main() {
match try_join!(task1(), task2()) {
Ok((res1, res2)) => {
println!("{}", res1);
println!("{}", res2);
}
Err(e) => println!("Error: {}", e),
}
}
在异步编程中，我们经常需要处理任务超时和取消的情况。Tokio 提供了相应的 API：
use tokio::time::timeout;
use std::time::Duration;
async fn long_running_task() -> String {
tokio::time::sleep(Duration::from_secs(3)).await;
"Task completed".to_string()
}
#[tokio::main]
async fn main() {
match timeout(Duration::from_secs(2), long_running_task()).await {
Ok(res) => println!("{}", res),
Err(e) => println!("Timeout: {}", e),
}
}
use tokio::select;
use std::time::Duration;
async fn task() -> String {
loop {
println!("Task running...");
tokio::time::sleep(Duration::from_secs(1)).await;
}
}
#[tokio::main]
async fn main() {
let mut interval = tokio::time::interval(Duration::from_secs(2));
select! {
res = task() => println!("{}", res),
_ = interval.tick() => println!("Task cancelled"),
}
}
异步函数可以返回 Result 类型，结合 ? 运算符进行错误处理：
use reqwest;
use std::io;
async fn fetch_data(url: &str) -> Result<String, Box<dyn std::error::Error>> {
let response = reqwest::get(url).await?;
let body = response.text().await?;
Ok(body)
}
async fn write_to_file(content: &str, filename: &str) -> Result<(), Box<dyn std::error::Error>> {
let mut file = tokio::fs::File::create(filename).await?;
tokio::io::AsyncWriteExt::write_all(&mut file, content.as_bytes()).await?;
Ok(())
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let url = "https://httpbin.org/get";
let content = fetch_data(url).await?;
write_to_file(&content, "data.txt").await?;
println!("Data saved to data.txt");
Ok(())
}
我们将构建一个高性能的 HTTP API 服务，提供用户的 CRUD 操作。项目需求如下：
项目架构设计：
创建项目：
cargo new rust-async-api
cd rust-async-api
在 Cargo.toml 中添加依赖：
[dependencies]
axum = "0.5"
tokio = { version = "1.0", features = ["full"] }
sqlx = { version = "0.6", features = ["postgres", "runtime-tokio-rustls"] }
serde = { version = "1.0", features = ["derive"] }
serde_json = "1.0"
tower-http = { version = "0.3", features = ["cors", "trace"] }
初始化 PostgreSQL 数据库：
CREATE DATABASE rust_async_api;
CREATE TABLE users (
id SERIAL PRIMARY KEY,
name VARCHAR(100) NOT NULL,
email VARCHAR(100) NOT NULL UNIQUE,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
创建 src/db.rs：
use sqlx::PgPool;
pub async fn create_pool(database_url: &str) -> PgPool {
PgPool::connect(database_url).await.unwrap()
}
创建 src/models.rs：
use serde::{Deserialize, Serialize};
use sqlx::FromRow;
#[derive(Debug, FromRow, Serialize, Deserialize)]
pub struct User {
pub id: i32,
pub name: String,
pub email: String,
pub created_at: chrono::DateTime<chrono::Utc>,
}
#[derive(Debug, Deserialize)]
pub struct CreateUser {
pub name: String,
pub email: String,
}
#[derive(Debug, Deserialize)]
pub struct UpdateUser {
pub name: Option<String>,
pub email: Option<String>,
}
创建 src/handlers.rs：
use axum::{extract::{Path, State}, http::StatusCode, response::IntoResponse, Json};
use sqlx::PgPool;
use crate::models::{CreateUser, UpdateUser, User};
pub async fn get_users(State(pool): State<PgPool>) -> impl IntoResponse {
let users = sqlx::query_as!(User, "SELECT * FROM users").fetch_all(&pool).await;
match users {
Ok(users) => (StatusCode::OK, Json(users)).into_response(),
Err(_) => (StatusCode::INTERNAL_SERVER_ERROR, "Failed to fetch users").into_response(),
}
}
pub async fn get_user(Path(user_id): Path<i32>, State(pool): State<PgPool>) -> impl IntoResponse {
let user = sqlx::query_as!(User, "SELECT * FROM users WHERE id = $1", user_id).fetch_one(&pool).await;
match user {
Ok(user) => (StatusCode::OK, Json(user)).into_response(),
Err(sqlx::Error::RowNotFound) => (StatusCode::NOT_FOUND, "User not found").into_response(),
Err(_) => (StatusCode::INTERNAL_SERVER_ERROR, "Failed to fetch user").into_response(),
}
}
pub async fn create_user(State(pool): State<PgPool>, Json(payload): Json<CreateUser>) -> impl IntoResponse {
let user = sqlx::query_as!(User, "INSERT INTO users (name, email) VALUES ($1, $2) RETURNING *", payload.name, payload.email).fetch_one(&pool).await;
match user {
Ok(user) => (StatusCode::CREATED, Json(user)).into_response(),
Err(sqlx::Error::Database(dbe)) if dbe.code().unwrap_or("").eq_ignore_ascii_case("23505") => {
(StatusCode::BAD_REQUEST, "Email already exists").into_response()
}
Err(_) => (StatusCode::INTERNAL_SERVER_ERROR, "Failed to create user").into_response(),
}
}
pub async fn update_user(Path(user_id): Path<i32>, State(pool): State<PgPool>, Json(payload): Json<UpdateUser>) -> impl IntoResponse {
let user = sqlx::query_as!(User, "SELECT * FROM users WHERE id = $1", user_id).fetch_one(&pool).await;
if user.is_err() {
return (StatusCode::NOT_FOUND, "User not found").into_response();
}
let user = user.unwrap();
let name = payload.name.unwrap_or(user.name);
let email = payload.email.unwrap_or(user.email);
let updated_user = sqlx::query_as!(User, "UPDATE users SET name = $1, email = $2 WHERE id = $3 RETURNING *", name, email, user_id).fetch_one(&pool).await;
match updated_user {
Ok(user) => (StatusCode::OK, Json(user)).into_response(),
Err(sqlx::Error::Database(dbe)) if dbe.code().unwrap_or("").eq_ignore_ascii_case("23505") => {
(StatusCode::BAD_REQUEST, "Email already exists").into_response()
}
Err(_) => (StatusCode::INTERNAL_SERVER_ERROR, "Failed to update user").into_response(),
}
}
pub async fn delete_user(Path(user_id): Path<i32>, State(pool): State<PgPool>) -> impl IntoResponse {
let result = sqlx::query!("DELETE FROM users WHERE id = $1", user_id).execute(&pool).await;
match result {
Ok(result) if result.rows_affected() == 0 => (StatusCode::NOT_FOUND, "User not found").into_response(),
Ok(_) => (StatusCode::NO_CONTENT, "").into_response(),
Err(_) => (StatusCode::INTERNAL_SERVER_ERROR, "Failed to delete user").into_response(),
}
}
创建 src/main.rs：
use axum::{routing::{get, post, put, delete}, Router};
use tower_http::{cors::CorsLayer, trace::TraceLayer};
use std::env;
use crate::db::create_pool;
use crate::handlers::{get_users, get_user, create_user, update_user, delete_user};
mod db;
mod handlers;
mod models;
#[tokio::main]
async fn main() {
// 初始化日志
tracing_subscriber::fmt::init();
// 从环境变量获取数据库 URL
let database_url = env::var("DATABASE_URL").expect("DATABASE_URL is required");
let pool = create_pool(&database_url).await;
// 配置 CORS
let cors = CorsLayer::permissive();
// 创建路由
let app = Router::new()
.route("/users", get(get_users))
.route("/users", post(create_user))
.route("/users/:id", get(get_user))
.route("/users/:id", put(update_user))
.route("/users/:id", delete(delete_user))
.layer(cors)
.layer(TraceLayer::new_for_http())
.with_state(pool);
// 启动服务器
let listener = tokio::net::TcpListener::bind("0.0.0.0:3000").await.unwrap();
println!("Server running on http://0.0.0.0:3000");
axum::serve(listener, app).await.unwrap();
}
使用 wrk 进行性能测试：
wrk -t12 -c400 -d30s http://localhost:3000/users
测试结果（示例）：
Running 30s test @ http://localhost:3000/users
12 threads and 400 connections
Thread Stats Avg Stdev Max +/- Stdev
Latency 50.19ms 20.12ms 200.00ms 68.00%
Req/Sec 666.67 80.00 800.00 80.00%
239999 requests in 30.01s, 50.00MB read
Requests/sec: 7997.00
Transfer/sec: 1.67MB
可以使用 Docker 容器化部署 API 服务。创建 Dockerfile：
FROM rust:1.60 as builder
WORKDIR /app
COPY Cargo.toml Cargo.lock ./
RUN cargo new --bin rust-async-api
WORKDIR /app/rust-async-api
COPY src ./src
RUN cargo build --release
FROM debian:buster-slim
RUN apt-get update && apt-get install -y ca-certificates && rm -rf /var/lib/apt/lists/*
COPY --from=builder /app/rust-async-api/target/release/rust-async-api /usr/local/bin/
EXPOSE 3000
CMD ["rust-async-api"]
构建并运行 Docker 容器：
docker build -t rust-async-api .
docker run -p 3000:3000 -e DATABASE_URL=postgresql://user:password@host:5432/rust_async_api rust-async-api
使用 Prometheus 和 Grafana 监控 API 服务的性能。添加 Prometheus 依赖：
[dependencies]
axum-prometheus = "0.4"
prometheus = "0.13"
在 src/main.rs 中添加 Prometheus 监控：
use axum_prometheus::PrometheusMetrics;
let prometheus = PrometheusMetrics::new("api", Some("/metrics"));
let app = Router::new()
// 其他路由
.route("/metrics", get(prometheus.metrics))
.layer(prometheus.layer())
// 其他中间件
.with_state(pool);
Rust 的异步编程提供了高性能、内存安全的并发处理能力。通过 async/await 语法，我们可以编写出与同步代码类似的异步代码，同时获得异步编程的优势。Tokio 作为成熟的异步运行时库，提供了丰富的 API 和组件，使开发高性能网络应用变得更加简单。
在实际项目中，我们需要注意异步编程的常见错误，遵循最佳实践，同时利用生态系统中的工具和库来提高开发效率和代码质量。希望本章的内容能够帮助您快速掌握 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