Rust 异步 Web 框架 Axum：核心原理与实战进阶

Rust 异步 Web 框架 Axum：核心原理与实战进阶

Axum 作为 Tokio 团队官方维护的 Rust Web 框架，凭借其类型安全和模块化设计，在高性能异步服务开发中占据了重要地位。它完全基于 Tokio 运行时，充分利用现代硬件的并发能力，同时提供了简单易用的 API。

架构与核心组件

设计理念

Axum 的核心在于模块化与可扩展性。通过中间件、请求提取器和响应映射器，开发者可以灵活组合功能。更重要的是，它利用 Rust 的类型系统确保请求处理逻辑的正确性，大幅减少运行时错误。

请求提取器

请求提取器负责从 HTTP 请求中提取数据，如路径参数或请求体。Axum 内置了多种提取器，也支持自定义。

内置提取器示例

use axum::{extract::Path, response::IntoResponse, routing::get, Router};

async fn get_user(Path(user_id): Path<i32>) -> impl IntoResponse {
    format!("Get user with ID: {}", user_id)
}

#[tokio::main]
async fn main() {
    let app = Router::new().route("/users/:id", get(get_user));
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

自定义提取器示例 如果需要提取特定 Header，比如 User-Agent，可以这样实现：

use axum::{async_trait, extract::FromRequestParts, http::request::Parts, response::IntoResponse, routing::get, Router};

struct UserAgent(String);

#[async_trait]
impl FromRequestParts<()> for UserAgent {
    type Rejection = ();
    async fn from_request_parts(parts: &mut Parts, _state: &()) -> Result<Self, Self::Rejection> {
        parts.headers.get("user-agent")
            .and_then(|value| value.to_str().ok())
            .map(|s| UserAgent(s.to_string()))
            .ok_or(())
    }
}

async fn get_user_agent(agent: UserAgent) -> impl IntoResponse {
    format!("User-Agent: {}", agent.0)
}

#[tokio::main]
async fn main() {
    let app = Router::new().route("/user-agent", get(get_user_agent));
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

响应映射器

响应映射器将函数返回值转换为 HTTP 响应。你可以直接返回状态码和 JSON，也可以自定义结构体。

内置映射器示例

use axum::{http::StatusCode, response::IntoResponse, routing::get, Router};
use serde_json::json;

async fn get_user() -> impl IntoResponse {
    (StatusCode::OK, json!({"id": 1, "name": "张三", "email": "[email protected]"}))
}

async fn create_user() -> impl IntoResponse {
    (StatusCode::CREATED, "User created successfully")
}

async fn delete_user() -> impl IntoResponse {
    StatusCode::NO_CONTENT
}

#[tokio::main]
async fn main() {
    let app = Router::new()
        .route("/users/1", get(get_user))
        .route("/users", get(create_user))
        .route("/users/1", get(delete_user));
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

自定义映射器示例 如果你希望统一返回格式，可以封装一个 ApiResponse 结构体：

use axum::{http::StatusCode, response::IntoResponse, routing::get, Router};
use serde_json::json;

struct ApiResponse {
    code: i32,
    message: String,
    data: serde_json::Value,
}

impl IntoResponse for ApiResponse {
    fn into_response(self) -> axum::response::Response {
        let status = if self.code == 200 { StatusCode::OK } else { StatusCode::BAD_REQUEST };
        (status, json!({"code": self.code, "message": self.message, "data": self.data})).into_response()
    }
}

async fn get_user() -> ApiResponse {
    ApiResponse {
        code: 200,
        message: "Success".to_string(),
        data: json!({"id": 1, "name": "张三", "email": "[email protected]"}),
    }
}

#[tokio::main]
async fn main() {
    let app = Router::new().route("/users/1", get(get_user));
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

中间件

中间件是处理请求和响应的通用组件，常用于身份验证、日志记录等。

内置中间件示例 使用 tower_http 可以轻松添加追踪层：

use axum::{middleware, routing::get, Router};
use tower_http::trace::TraceLayer;

async fn handler() -> &'static str { "Hello, World!" }

#[tokio::main]
async fn main() {
    let app = Router::new()
        .route("/", get(handler))
        .layer(TraceLayer::new_for_http());
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

自定义中间件示例 如果需要计算响应时间，可以编写自定义中间件：

use axum::{async_trait, extract::FromRequestParts, http::request::Parts, middleware::Next, response::IntoResponse, routing::get, Router};
use std::time::Duration;
use tokio::time::Instant;

struct RequestTime(Duration);

#[async_trait]
impl FromRequestParts<()> for RequestTime {
    type Rejection = ();
    async fn from_request_parts(parts: &mut Parts, _state: &()) -> Result<Self, Self::Rejection> {
        parts.extensions.get::<RequestTime>().copied().ok_or(())
    }
}

async fn timing_middleware<B>(request: axum::http::Request<B>, next: Next<B>) -> impl IntoResponse {
    let start = Instant::now();
    let response = next.run(request).await;
    let duration = start.elapsed();
    let mut response = response.into_response();
    response.headers().insert("X-Response-Time", format!("{}ms", duration.as_millis()).parse().unwrap());
    response
}

async fn handler(time: RequestTime) -> impl IntoResponse {
    format!("Response time: {}ms", time.0.as_millis())
}

#[tokio::main]
async fn main() {
    let app = Router::new()
        .route("/", get(handler))
        .layer(middleware::from_fn(timing_middleware));
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

路由系统详解

定义与匹配

Axum 的路由基于路径匹配，支持静态、动态参数和通配符。

静态与动态路径

use axum::{extract::Path, response::IntoResponse, routing::get, Router};

async fn get_user(Path(user_id): Path<i32>) -> impl IntoResponse {
    format!("Get user with ID: {}", user_id)
}

async fn get_product(Path((category_id, product_id)): Path<(i32, i32)>) -> impl IntoResponse {
    format!("Get product {} in category {}", product_id, category_id)
}

#[tokio::main]
async fn main() {
    let app = Router::new()
        .route("/users/:id", get(get_user))
        .route("/categories/:category_id/products/:product_id", get(get_product));
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

通配符路径 用于捕获未匹配的路径，通常配合 404 处理：

use axum::{extract::Path, response::IntoResponse, routing::get, Router};

async fn catch_all(Path(path): Path<String>) -> impl IntoResponse {
    format!("Not found: {}", path)
}

#[tokio::main]
async fn main() {
    let app = Router::new().route("/:rest..", get(catch_all));
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

嵌套与组合

将相关路由组织成模块，能提高代码可读性。

路由嵌套

use axum::{extract::Path, response::IntoResponse, routing::{get, post}, Router};

async fn get_user(Path(user_id): Path<i32>) -> impl IntoResponse {
    format!("Get user with ID: {}", user_id)
}

async fn create_user() -> impl IntoResponse { "User created successfully" }

async fn delete_user(Path(user_id): Path<i32>) -> impl IntoResponse {
    format!("Delete user with ID: {}", user_id)
}

#[tokio::main]
async fn main() {
    let user_routes = Router::new()
        .route("/:id", get(get_user).delete(delete_user))
        .route("/", post(create_user));
    let app = Router::new().nest("/users", user_routes);
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

路由组合 合并多个路由组：

use axum::{response::IntoResponse, routing::get, Router};

async fn home() -> impl IntoResponse { "Home page" }
async fn about() -> impl IntoResponse { "About page" }
async fn contact() -> impl IntoResponse { "Contact page" }

#[tokio::main]
async fn main() {
    let public_routes = Router::new().route("/", get(home)).route("/about", get(about));
    let contact_routes = Router::new().route("/contact", get(contact));
    let app = public_routes.merge(contact_routes);
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

状态共享

通过 Router.with_state 共享数据库连接池、配置等信息。

use axum::{extract::State, response::IntoResponse, routing::get, Router};
use sqlx::PgPool;
use std::sync::Arc;

#[derive(Clone)]
struct AppState {
    db_pool: Arc<PgPool>,
    config: crate::config::Config,
}

async fn get_user_count(State(state): State<AppState>) -> impl IntoResponse {
    let count = sqlx::query_scalar!("SELECT COUNT(*) FROM users")
        .fetch_one(&*state.db_pool)
        .await
        .unwrap();
    format!("Total users: {}", count)
}

#[tokio::main]
async fn main() {
    let config = crate::config::Config::from_env().unwrap();
    let db_pool = Arc::new(sqlx::PgPool::connect(&config.db.url).await.unwrap());
    let state = AppState { db_pool, config };
    let app = Router::new()
        .route("/users/count", get(get_user_count))
        .with_state(state);
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

高级功能实战

WebSocket 支持

Axum 原生支持 WebSocket，适合实时通信场景。

use axum::{extract::WebSocketUpgrade, response::IntoResponse, routing::get, Router};
use tokio_tungstenite::tungstenite::Message;

async fn websocket_handler(ws: WebSocketUpgrade) -> impl IntoResponse {
    ws.on_upgrade(|mut socket| async move {
        println!("WebSocket connection established");
        while let Some(msg) = socket.next().await {
            match msg {
                Ok(Message::Text(text)) => {
                    println!("Received text message: {}", text);
                    socket.send(Message::Text(format!("Echo: {}", text))).await.unwrap();
                }
                Ok(Message::Binary(data)) => {
                    println!("Received binary message with length: {}", data.len());
                    socket.send(Message::Binary(data)).await.unwrap();
                }
                Ok(Message::Ping(ping)) => {
                    socket.send(Message::Pong(ping)).await.unwrap();
                }
                Ok(Message::Pong(_)) => {}
                Ok(Message::Close(frame)) => {
                    println!("WebSocket connection closing: {:?}", frame);
                }
                Err(e) => {
                    println!("WebSocket error: {:?}", e);
                }
            }
        }
        println!("WebSocket connection closed");
    })
}

#[tokio::main]
async fn main() {
    let app = Router::new().route("/ws", get(websocket_handler));
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

流式请求与响应

适用于处理大量数据的场景。

流式响应

use axum::{body::Body, response::IntoResponse, routing::get, Router};
use futures::stream::{self, StreamExt};
use http_body_util::Full;

async fn stream_response() -> impl IntoResponse {
    let items = vec!["First item", "Second item", "Third item"];
    let stream = stream::iter(items).map(|item| Ok::<_, std::io::Error>(Full::new(item.as_bytes())));
    Body::wrap_stream(stream)
}

#[tokio::main]
async fn main() {
    let app = Router::new().route("/stream", get(stream_response));
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

流式请求

use axum::{body::Body, extract::Request, response::IntoResponse, routing::post, Router};
use futures::StreamExt;

async fn stream_request(request: Request<Body>) -> impl IntoResponse {
    let mut body = request.into_body();
    let mut buffer = Vec::new();
    while let Some(chunk) = body.next().await {
        buffer.extend_from_slice(&chunk.unwrap());
    }
    format!("Received {} bytes", buffer.len())
}

#[tokio::main]
async fn main() {
    let app = Router::new().route("/upload", post(stream_request));
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

错误处理与响应

灵活的错误处理机制允许自定义错误类型和响应格式。

use axum::{extract::Path, http::StatusCode, response::{IntoResponse, Response}, routing::get, Router};
use serde_json::json;
use thiserror::Error;

#[derive(Error, Debug)]
enum AppError {
    #[error("User not found")]
    UserNotFound,
    #[error("Invalid request")]
    InvalidRequest,
    #[error(transparent)]
    Unexpected(#[from] anyhow::Error),
}

impl IntoResponse for AppError {
    fn into_response(self) -> Response {
        let (status, message) = match self {
            AppError::UserNotFound => (StatusCode::NOT_FOUND, "User not found"),
            AppError::InvalidRequest => (StatusCode::BAD_REQUEST, "Invalid request"),
            AppError::Unexpected(_) => (StatusCode::INTERNAL_SERVER_ERROR, "Internal server error"),
        };
        (status, json!({"code": status.as_u16(), "message": message})).into_response()
    }
}

async fn get_user(Path(user_id): Path<i32>) -> Result<impl IntoResponse, AppError> {
    if user_id == 0 {
        return Err(AppError::InvalidRequest);
    }
    if user_id == 999 {
        return Err(AppError::UserNotFound);
    }
    Ok(json!({"id": user_id, "name": "张三", "email": "[email protected]"}))
}

#[tokio::main]
async fn main() {
    let app = Router::new().route("/users/:id", get(get_user));
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

CORS 支持

通过 tower_http 的 cors 中间件轻松配置跨域规则。

use axum::{response::IntoResponse, routing::get, Router};
use tower_http::cors::{Any, CorsLayer};

async fn handler() -> impl IntoResponse { "CORS enabled" }

#[tokio::main]
async fn main() {
    let cors = CorsLayer::new()
        .allow_origin(Any)
        .allow_methods(Any)
        .allow_headers(Any);
    let app = Router::new()
        .route("/", get(handler))
        .layer(cors);
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

身份验证与授权

支持 JWT、API 密钥等多种方式。

JWT 示例

use axum::{async_trait, extract::FromRequestParts, http::request::Parts, response::IntoResponse, routing::{get, post}, Router};
use jsonwebtoken::{decode, encode, Algorithm, DecodingKey, EncodingKey, Header, Validation};
use serde::{Deserialize, Serialize};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

#[derive(Debug, Serialize, Deserialize)]
struct Claims {
    sub: String,
    exp: usize,
}

impl Claims {
    fn new(sub: &str) -> Self {
        let expiration = SystemTime::now()
            .checked_add(Duration::from_secs(3600))
            .unwrap()
            .duration_since(UNIX_EPOCH)
            .unwrap()
            .as_secs() as usize;
        Claims { sub: sub.to_string(), exp: expiration }
    }
}

struct JwtSecret(String);

#[async_trait]
impl FromRequestParts<JwtSecret> for Claims {
    type Rejection = ();
    async fn from_request_parts(parts: &mut Parts, state: &JwtSecret) -> Result<Self, Self::Rejection> {
        parts.headers.get("authorization")
            .and_then(|value| value.to_str().ok())
            .and_then(|s| s.strip_prefix("Bearer ").map(|s| s.to_string()))
            .and_then(|token| {
                decode::<Claims>(&token, &DecodingKey::from_secret(state.0.as_bytes()), &Validation::new(Algorithm::HS256)).ok()
            })
            .map(|data| data.claims)
            .ok_or(())
    }
}

async fn login() -> impl IntoResponse {
    let claims = Claims::new("user123");
    let token = encode(&Header::new(Algorithm::HS256), &claims, &EncodingKey::from_secret(b"secret")).unwrap();
    token
}

async fn protected(claims: Claims) -> impl IntoResponse {
    format!("Welcome, {}", claims.sub)
}

#[tokio::main]
async fn main() {
    let secret = JwtSecret("secret".to_string());
    let public_routes = Router::new().route("/login", post(login));
    let protected_routes = Router::new().route("/protected", get(protected)).with_state(secret.clone());
    let app = public_routes.merge(protected_routes);
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

性能优化建议

工作线程数配置

Axum 依赖 Tokio，合理配置工作线程数能提升并发能力。

use axum::{response::IntoResponse, routing::get, Router};
use num_cpus;
use tokio::runtime::Builder;

async fn handler() -> impl IntoResponse { "Hello, World!" }

fn main() {
    let runtime = Builder::new_multi_thread()
        .worker_threads(num_cpus::get())
        .max_blocking_threads(10)
        .build()
        .unwrap();
    runtime.block_on(async {
        let app = Router::new().route("/", get(handler));
        axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
            .serve(app.into_make_service())
            .await
            .unwrap();
    });
}

提取器与映射器优化

避免重复解析数据，利用 FromRef 和 FromRequestParts 优化状态访问。

use axum::{async_trait, extract::{FromRef, FromRequestParts}, http::request::Parts, response::IntoResponse, routing::get, Router};
use serde_json::json;
use std::sync::Arc;

#[derive(Clone)]
struct AppState {
    db_pool: Arc<sqlx::PgPool>,
    config: crate::config::Config,
}

struct UserExtractor(i32);

#[async_trait]
impl FromRef<AppState> for crate::db::DbPool {
    fn from_ref(state: &AppState) -> Self {
        state.db_pool.clone()
    }
}

#[async_trait]
impl FromRequestParts<AppState> for UserExtractor {
    type Rejection = ();
    async fn from_request_parts(parts: &mut Parts, state: &AppState) -> Result<Self, Self::Rejection> {
        parts.headers.get("user-id")
            .and_then(|value| value.to_str().ok())
            .and_then(|s| s.parse().ok())
            .map(|id| UserExtractor(id))
            .ok_or(())
    }
}

async fn get_user(extractor: UserExtractor) -> impl IntoResponse {
    json!({"id": extractor.0, "name": "张三", "email": "[email protected]"})
}

#[tokio::main]
async fn main() {
    let config = crate::config::Config::from_env().unwrap();
    let db_pool = Arc::new(sqlx::PgPool::connect(&config.db.url).await.unwrap());
    let state = AppState { db_pool, config };
    let app = Router::new()
        .route("/users", get(get_user))
        .with_state(state);
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

中间件优化

使用 tower_http 的内置中间件可以减少开销。

use axum::{response::IntoResponse, routing::get, Router};
use tower_http::trace::{DefaultMakeSpan, DefaultOnResponse, TraceLayer};

async fn handler() -> impl IntoResponse { "Hello, World!" }

#[tokio::main]
async fn main() {
    let app = Router::new()
        .route("/", get(handler))
        .layer(TraceLayer::new_for_http()
            .make_span_with(DefaultMakeSpan::new().include_headers(true))
            .on_response(DefaultOnResponse::new().include_headers(true)));
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

实战项目集成

用户同步服务

// user-sync-service/src/main.rs
use axum::{http::StatusCode, response::IntoResponse, routing::{get, post}, Router};
use user_sync_service::sync;
use user_sync_service::config::Config;

async fn health() -> impl IntoResponse {
    StatusCode::OK
}

async fn sync_users() -> impl IntoResponse {
    match sync::sync_users().await {
        Ok(_) => StatusCode::ACCEPTED,
        Err(e) => {
            tracing::error!("User sync failed: {:?}", e);
            StatusCode::INTERNAL_SERVER_ERROR
        }
    }
}

#[tokio::main]
async fn main() {
    let config = Config::from_env().unwrap();
    let app = Router::new()
        .route("/health", get(health))
        .route("/api/users/sync", post(sync_users));
    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

订单处理服务

// order-processing-service/src/main.rs
use axum::{http::StatusCode, response::IntoResponse, routing::{get, post}, Router};
use order_processing_service::process;
use order_processing_service::config::Config;

async fn health() -> impl IntoResponse {
    StatusCode::OK
}

async fn process_order() -> impl IntoResponse {
    match process::process_orders().await {
        Ok(_) => StatusCode::ACCEPTED,
        Err(e) => {
            tracing::error!("Order processing failed: {:?}", e);
            StatusCode::INTERNAL_SERVER_ERROR
        }
    }
}

#[tokio::main]
async fn main() {
    let config = Config::from_env().unwrap();
    let app = Router::new()
        .route("/health", get(health))
        .route("/api/orders/process", post(process_order));
    axum::Server::bind(&"0.0.0.0:3001".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

监控服务

// monitoring-service/src/main.rs
use axum::{extract::WebSocketUpgrade, http::StatusCode, response::IntoResponse, routing::{get, post}, Router};
use monitoring_service::monitor;
use monitoring_service::config::Config;

async fn health() -> impl IntoResponse {
    StatusCode::OK
}

async fn websocket_handler(ws: WebSocketUpgrade) -> impl IntoResponse {
    monitor::handle_websocket_connection(ws).await
}

#[tokio::main]
async fn main() {
    let config = Config::from_env().unwrap();
    let app = Router::new()
        .route("/health", get(health))
        .route("/ws", get(websocket_handler));
    axum::Server::bind(&"0.0.0.0:3002".parse().unwrap())
        .serve(app.into_make_service())
        .await
        .unwrap();
}

常见问题与解决方案

请求提取器类型不匹配

当提取器类型与请求数据不一致时，会报错。确保路径参数 :id 的类型（如 i32）与实际传入值匹配。

响应映射器返回值不匹配

返回值必须实现 IntoResponse trait。建议使用 (StatusCode, json) 或自定义类型。

中间件顺序问题

中间件顺序影响处理流程。身份验证应放在路由之前，CORS 通常在最外层。

状态共享生命周期

确保状态类型实现了 Clone，并使用 Arc 管理共享数据，避免编译错误或运行时崩溃。

总结

Axum 是一个功能强大且易用的异步 Web 框架，依托 Tokio 运行时，具备高度模块化和类型安全特性。掌握其架构、路由、高级功能及性能优化方法，能帮助开发者构建高质量的异步 Web 应用。