Rust
Rust 异步 API 设计与 Axum 框架实战
本文介绍了 Rust 异步 API 的设计原则，包括简单性、一致性、可扩展性、安全性和性能。详细讲解了 Axum 框架的核心功能，如路由设计、中间件、错误处理、参数提取和响应构建。通过实战项目集成，展示了如何构建包含认证、缓存、消息队列的用户管理系统，并提供了性能优化及部署监控方案。
本文介绍了 Rust 异步 API 的设计原则，包括简单性、一致性、可扩展性、安全性和性能。详细讲解了 Axum 框架的核心功能，如路由设计、中间件、错误处理、参数提取和响应构建。通过实战项目集成，展示了如何构建包含认证、缓存、消息队列的用户管理系统，并提供了性能优化及部署监控方案。
💡异步 API 是现代 Web 应用和微服务架构中的重要组成部分，通过非阻塞性的请求处理，显著提高了系统的并发能力和响应速度。Rust 语言的异步特性（基于 Tokio 运行时）和 Axum 框架的简洁设计，使得构建高性能、可靠的异步 API 变得更加容易。
Axum 是一个由 Tokio 团队开发的异步 Web 框架，具有以下特点：
本章将深入探讨异步 API 的设计原则，介绍 Axum 框架的核心功能，包括路由设计、中间件、错误处理、参数提取和响应构建，并通过实战项目集成演示如何构建高性能的异步 API。
API 设计应该简洁明了，易于理解和使用。避免过于复杂的接口和参数，提高开发效率和可维护性。
好的 API 设计：
// 获取用户信息
#[get("/api/users/:id")]
async fn get_user(id: Path<i32>) -> impl IntoResponse {
// 业务逻辑
}
// 分页获取用户列表
#[get("/api/users")]
async fn get_users(page: Option<u32>, page_size: Option<u32>) -> impl IntoResponse {
// 业务逻辑
}
差的 API 设计：
// 获取用户信息
#[get("/api/getuserbyid?id=:id")]
async fn get_user(id: Query<i32>) -> impl IntoResponse {
// 业务逻辑
}
// 分页获取用户列表
#[get("/api/getusers?page=:page&pagesize=:page_size&sortby=:sort_by&sortorder=:sort_order")]
async fn get_users(
page: Query<u32>,
page_size: Query<u32>,
sort_by: Query<String>,
sort_order: Query<String>
) -> impl IntoResponse {
// 业务逻辑
}
API 设计应该遵循一致的命名规范、参数格式和响应格式，提高 API 的可读性和可维护性。
命名规范：
参数格式：
响应格式：
统一的错误格式，如：
{"error":"Resource not found","message":"The user with ID 123 was not found","code":404}
API 设计应该考虑未来的扩展需求，避免频繁的 API 版本升级。可以使用版本化 API、参数可选性和扩展字段来提高 API 的可扩展性。
版本化 API：
// 第一个版本的 API
#[get("/api/v1/users/:id")]
async fn get_user_v1(id: Path<i32>) -> impl IntoResponse {
// 业务逻辑
}
// 第二个版本的 API
#[get("/api/v2/users/:id")]
async fn get_user_v2(id: Path<i32>) -> impl IntoResponse {
// 业务逻辑
}
参数可选性：
// 分页获取用户列表，page 和 page_size 为可选参数
#[get("/api/users")]
async fn get_users(page: Option<u32>, page_size: Option<u32>) -> impl IntoResponse {
let page = page.unwrap_or(1);
let page_size = page_size.unwrap_or(10);
// 业务逻辑
}
扩展字段：
// 用户信息响应，包含扩展字段
#[derive(Debug, Serialize, Deserialize)]
struct UserResponse {
id: i32,
name: String,
email: String,
#[serde(skip_serializing_if = "Option::is_none")]
phone: Option<String>,
#[serde(flatten)]
extra_fields: BTreeMap<String, Value>,
}
API 设计应该考虑安全需求，如身份认证、授权、数据加密和防止攻击。可以使用 JWT、OAuth2、HTTPS 和 CORS 等安全机制。
身份认证：
// JWT 身份认证中间件
async fn jwt_auth<B>(
headers: HeaderMap,
req: Request<B>,
next: Next<B>,
) -> Result<Response, JwtError> {
// 从请求头中获取 JWT token
let token = headers.get("Authorization").and_then(|value| value.to_str().ok()).and_then(|s| s.strip_prefix("Bearer "));
if let Some(token) = token {
// 验证 JWT token
let claims = decode_token(token)?;
// 将用户信息存储在请求扩展中
let mut req = req;
req.extensions_mut().insert(claims);
Ok(next.run(req).await)
} else {
Err(JwtError::MissingToken)
}
}
授权：
// 授权中间件，检查用户是否有访问资源的权限
async fn authorize<B>(
claims: Extension<Claims>,
req: Request<B>,
next: Next<B>,
) -> Result<Response, AuthorizationError> {
// 获取资源类型和操作类型
let resource_type = req.uri().path().split('/').nth(2).unwrap_or("");
let operation_type = req.method().as_str().to_lowercase();
// 检查用户权限
if claims.has_permission(resource_type, &operation_type) {
Ok(next.run(req).await)
} else {
Err(AuthorizationError::Forbidden)
}
}
API 设计应该考虑性能需求，如减少请求响应时间、提高并发能力和优化资源消耗。可以使用缓存、异步处理和数据压缩等性能优化方法。
缓存：
// 数据缓存中间件
async fn cache<B>(
req: Request<B>,
next: Next<B>,
) -> Result<Response, CacheError> {
let cache_key = format!("{}_{}", req.method(), req.uri().path());
if let Some(cached) = crate::cache::get(&cache_key).await {
Ok(Response::builder()
.status(StatusCode::OK)
.header("Content-Type", "application/json")
.body(Body::from(cached))
.unwrap())
} else {
let response = next.run(req).await;
if response.status().is_success() {
let body = hyper::body::to_bytes(response.into_body()).await?;
crate::cache::put(&cache_key, body.to_vec(), Duration::from_secs(60)).await?;
Ok(Response::builder()
.status(StatusCode::OK)
.header("Content-Type", "application/json")
.body(Body::from(body))
.unwrap())
} else {
Ok(response)
}
}
}
Axum 的路由设计基于 HTTP 方法和路径，使用宏定义路由，支持参数提取和中间件。
基本路由：
use axum::{routing::{get, post, put, delete}, Router};
async fn health() -> impl IntoResponse { "Healthy" }
async fn get_user() -> impl IntoResponse { "Get user" }
async fn create_user() -> impl IntoResponse { "Create user" }
async fn update_user() -> impl IntoResponse { "Update user" }
async fn delete_user() -> impl IntoResponse { "Delete user" }
#[tokio::main]
async fn main() {
let app = Router::new()
.route("/health", get(health))
.route("/api/users/:id", get(get_user))
.route("/api/users", post(create_user))
.route("/api/users/:id", put(update_user))
.route("/api/users/:id", delete(delete_user));
axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
.serve(app.into_make_service())
.await
.unwrap();
}
嵌套路由：
use axum::{routing::{get, post, put, delete}, Router};
async fn get_user() -> impl IntoResponse { "Get user" }
async fn create_user() -> impl IntoResponse { "Create user" }
async fn update_user() -> impl IntoResponse { "Update user" }
async fn delete_user() -> impl IntoResponse { "Delete user" }
async fn get_order() -> impl IntoResponse { "Get order" }
async fn create_order() -> impl IntoResponse { "Create order" }
async fn update_order() -> impl IntoResponse { "Update order" }
async fn delete_order() -> impl IntoResponse { "Delete order" }
#[tokio::main]
async fn main() {
let user_routes = Router::new()
.route("/:id", get(get_user))
.route("", post(create_user))
.route("/:id", put(update_user))
.route("/:id", delete(delete_user));
let order_routes = Router::new()
.route("/:id", get(get_order))
.route("", post(create_order))
.route("/:id", put(update_order))
.route("/:id", delete(delete_order));
let app = Router::new()
.nest("/api/users", user_routes)
.nest("/api/orders", order_routes);
axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
.serve(app.into_make_service())
.await
.unwrap();
}
Axum 的中间件用于在请求到达路由处理函数之前或之后执行逻辑，如身份认证、授权、日志记录和错误处理。
自定义中间件：
use axum::{extract::{Request, Response}, middleware::{self, Next}, routing::get, Router};
use tracing::info;
async fn log_request<B>(req: Request<B>, next: Next<B>) -> Response {
info!("Request: {} {:?} {:?}", req.method(), req.uri(), req.version());
let response = next.run(req).await;
info!("Response: {}", response.status());
response
}
async fn health() -> impl IntoResponse { "Healthy" }
#[tokio::main]
async fn main() {
let app = Router::new()
.route("/health", get(health))
.layer(middleware::from_fn(log_request));
axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
.serve(app.into_make_service())
.await
.unwrap();
}
第三方中间件：
use axum::{routing::get, Router};
use tower_http::cors::CorsLayer;
use tower_http::compression::CompressionLayer;
async fn health() -> impl IntoResponse { "Healthy" }
#[tokio::main]
async fn main() {
let app = Router::new()
.route("/health", get(health))
.layer(CorsLayer::permissive())
.layer(CompressionLayer::new());
axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
.serve(app.into_make_service())
.await
.unwrap();
}
Axum 的错误处理机制使用自定义错误类型和错误处理中间件，提供友好的错误信息。
自定义错误类型：
use thiserror::Error;
use axum::{http::StatusCode, response::{IntoResponse, Response}};
#[derive(Error, Debug)]
enum ApiError {
#[error("Resource not found")]
NotFound,
#[error("Invalid request: {0}")]
InvalidRequest(String),
#[error("Internal server error: {0}")]
InternalServerError(String),
}
impl IntoResponse for ApiError {
fn into_response(self) -> Response {
let (status, message) = match self {
ApiError::NotFound => (StatusCode::NOT_FOUND, "Resource not found"),
ApiError::InvalidRequest(msg) => (StatusCode::BAD_REQUEST, &msg),
ApiError::InternalServerError(msg) => (StatusCode::INTERNAL_SERVER_ERROR, &msg),
};
(status, message).into_response()
}
}
async fn get_user() -> Result<impl IntoResponse, ApiError> {
Err(ApiError::NotFound)
}
#[tokio::main]
async fn main() {
let app = Router::new().route("/api/users/:id", get(get_user));
axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
.serve(app.into_make_service())
.await
.unwrap();
}
全局错误处理：
use thiserror::Error;
use axum::{http::StatusCode, response::{IntoResponse, Response}, routing::get, Router};
#[derive(Error, Debug)]
enum ApiError {
#[error("Resource not found")]
NotFound,
#[error("Invalid request: {0}")]
InvalidRequest(String),
#[error("Internal server error: {0}")]
InternalServerError(String),
}
impl IntoResponse for ApiError {
fn into_response(self) -> Response {
let (status, message) = match self {
ApiError::NotFound => (StatusCode::NOT_FOUND, "Resource not found"),
ApiError::InvalidRequest(msg) => (StatusCode::BAD_REQUEST, &msg),
ApiError::InternalServerError(msg) => (StatusCode::INTERNAL_SERVER_ERROR, &msg),
};
(status, message).into_response()
}
}
async fn get_user() -> Result<impl IntoResponse, ApiError> {
Err(ApiError::NotFound)
}
async fn create_user() -> Result<impl IntoResponse, ApiError> {
Err(ApiError::InvalidRequest("Missing required field: name".to_string()))
}
async fn handle_error(err: ApiError) -> impl IntoResponse {
err.into_response()
}
#[tokio::main]
async fn main() {
let app = Router::new()
.route("/api/users/:id", get(get_user))
.route("/api/users", post(create_user))
.layer(axum::error_handling::HandleErrorLayer::new(handle_error));
axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
.serve(app.into_make_service())
.await
.unwrap();
}
Axum 的参数提取器用于从请求中提取参数，如路径参数、查询参数、请求体和请求头。
路径参数：
use axum::{extract::Path, routing::get, Router};
async fn get_user(Path(id): Path<i32>) -> impl IntoResponse {
format!("User ID: {}", id)
}
#[tokio::main]
async fn main() {
let app = Router::new().route("/api/users/:id", get(get_user));
axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
.serve(app.into_make_service())
.await
.unwrap();
}
查询参数：
use axum::{extract::Query, routing::get, Router};
use serde::Deserialize;
#[derive(Deserialize)]
struct PageParams {
page: Option<u32>,
page_size: Option<u32>,
}
async fn get_users(Query(params): Query<PageParams>) -> impl IntoResponse {
let page = params.page.unwrap_or(1);
let page_size = params.page_size.unwrap_or(10);
format!("Page: {}, Page Size: {}", page, page_size)
}
#[tokio::main]
async fn main() {
let app = Router::new().route("/api/users", get(get_users));
axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
.serve(app.into_make_service())
.await
.unwrap();
}
请求体：
use axum::{extract::Json, routing::post, Router};
use serde::Deserialize;
#[derive(Deserialize)]
struct CreateUserRequest {
name: String,
email: String,
phone: Option<String>,
}
async fn create_user(Json(req): Json<CreateUserRequest>) -> impl IntoResponse {
format!("Created user: {} ({})", req.name, req.email)
}
#[tokio::main]
async fn main() {
let app = Router::new().route("/api/users", post(create_user));
axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
.serve(app.into_make_service())
.await
.unwrap();
}
Axum 的响应构建器用于构建响应，支持 JSON、HTML、二进制数据和文件响应。
JSON 响应：
use axum::{extract::Path, routing::get, Router, Json};
use serde::Serialize;
#[derive(Serialize)]
struct UserResponse {
id: i32,
name: String,
email: String,
phone: Option<String>,
}
async fn get_user(Path(id): Path<i32>) -> impl IntoResponse {
let user = UserResponse {
id,
name: "张三".to_string(),
email: "[email protected]".to_string(),
phone: Some("13800138000".to_string()),
};
Json(user)
}
#[tokio::main]
async fn main() {
let app = Router::new().route("/api/users/:id", get(get_user));
axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
.serve(app.into_make_service())
.await
.unwrap();
}
HTML 响应：
use axum::{routing::get, Router, response::Html};
async fn get_index() -> impl IntoResponse {
Html(r#"
<!DOCTYPE html>
<html>
<head><title>Index</title></head>
<body><h1>Hello, Axum!</h1></body>
</html>
"#)
}
#[tokio::main]
async fn main() {
let app = Router::new().route("/", get(get_index));
axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
.serve(app.into_make_service())
.await
.unwrap();
}
文件响应：
use axum::{routing::get, Router, response::IntoResponse};
use std::fs::File;
use std::io::Read;
async fn get_file() -> impl IntoResponse {
let mut file = File::open("static/avatar.png").unwrap();
let mut buf = Vec::new();
file.read_to_end(&mut buf).unwrap();
(
axum::http::StatusCode::OK,
axum::http::HeaderMap::from_iter(vec![("Content-Type".to_string(), "image/png".to_string())]),
buf,
)
}
#[tokio::main]
async fn main() {
let app = Router::new().route("/static/avatar.png", get(get_file));
axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
.serve(app.into_make_service())
.await
.unwrap();
}
我们将使用 Axum 框架构建一个用户管理系统的异步 API，包括用户的 CRUD 操作、身份认证、授权和数据验证。系统架构如下：
项目结构如下：
user-management-api/
├── src/
│ ├── api/
│ │ ├── auth.rs
│ │ ├── user.rs
│ │ └── mod.rs
│ ├── config/
│ │ └── mod.rs
│ ├── db/
│ │ ├── user_repo.rs
│ │ └── mod.rs
│ ├── cache/
│ │ └── mod.rs
│ ├── mq/
│ │ └── mod.rs
│ ├── middleware/
│ │ ├── auth.rs
│ │ ├── cors.rs
│ │ ├── compression.rs
│ │ └── mod.rs
│ ├── error.rs
│ ├── main.rs
│ └── lib.rs
├── static/
│ └── avatar.png
├── Cargo.toml
└── Cargo.lock
配置管理使用 config 库，支持从环境变量和配置文件中读取配置。
配置结构：
// src/config/mod.rs
use config::Config;
use serde::Deserialize;
#[derive(Debug, Deserialize)]
pub struct DatabaseConfig {
pub url: String,
}
#[derive(Debug, Deserialize)]
pub struct RedisConfig {
pub url: String,
}
#[derive(Debug, Deserialize)]
pub struct RabbitMQConfig {
pub uri: String,
pub email_queue: String,
}
#[derive(Debug, Deserialize)]
pub struct JwtConfig {
pub secret: String,
pub expiry: i64,
}
#[derive(Debug, Deserialize)]
pub struct ServerConfig {
pub address: String,
pub port: u16,
}
#[derive(Debug, Deserialize)]
pub struct AppConfig {
pub database: DatabaseConfig,
pub redis: RedisConfig,
pub rabbitmq: RabbitMQConfig,
pub jwt: JwtConfig,
pub server: ServerConfig,
}
impl AppConfig {
pub fn from_env() -> Result<Self, config::ConfigError> {
Config::builder()
.add_source(config::Environment::default().separator("__"))
.add_source(config::File::with_name("config"))
.build()?.try_deserialize()
}
}
数据库访问使用 SQLx 库，连接到 PostgreSQL 数据库，执行用户的 CRUD 操作。
用户仓库：
// src/db/user_repo.rs
use crate::config::DatabaseConfig;
use crate::db::DbPool;
use crate::error::ApiError;
use crate::api::user::*;
use sqlx::FromRow;
use chrono::prelude::*;
#[derive(Debug, FromRow)]
struct User {
id: i32,
name: String,
email: String,
phone: Option<String>,
created_at: NaiveDateTime,
updated_at: NaiveDateTime,
}
impl From<User> for UserResponse {
fn from(user: User) -> Self {
UserResponse {
id: user.id,
name: user.name,
email: user.email,
phone: user.phone,
created_at: user.created_at,
updated_at: user.updated_at,
}
}
}
pub struct UserRepository {
pool: DbPool,
}
impl UserRepository {
pub fn new(config: &DatabaseConfig) -> Self {
let pool = sqlx::postgres::PgPoolOptions::new()
.max_connections(5)
.connect_lazy(&config.url)
.unwrap();
UserRepository { pool }
}
pub async fn get_user(&self, id: i32) -> Result<Option<UserResponse>, ApiError> {
let user = sqlx::query_as::<_, User>("SELECT * FROM users WHERE id = $1")
.bind(id)
.fetch_optional(&self.pool)
.await?;
Ok(user.map(|u| u.into()))
}
pub async fn get_users(&self, page: u32, page_size: u32) -> Result<Vec<UserResponse>, ApiError> {
let offset = (page - 1) * page_size;
let users = sqlx::query_as::<_, User>("SELECT * FROM users LIMIT $1 OFFSET $2")
.bind(page_size as i64)
.bind(offset as i64)
.fetch_all(&self.pool)
.await?;
Ok(users.into_iter().map(|u| u.into()).collect())
}
pub async fn create_user(&self, req: CreateUserRequest) -> Result<UserResponse, ApiError> {
let user = sqlx::query_as::<_, User>("INSERT INTO users (name, email, phone) VALUES ($1, $2, $3) RETURNING *",
).bind(&req.name).bind(&req.email).bind(&req.phone).fetch_one(&self.pool).await?;
Ok(user.into())
}
pub async fn update_user(&self, id: i32, req: UpdateUserRequest) -> Result<UserResponse, ApiError> {
let user = sqlx::query_as::<_, User>("UPDATE users SET name = $1, email = $2, phone = $3, updated_at = NOW() WHERE id = $4 RETURNING *",
).bind(&req.name).bind(&req.email).bind(&req.phone).bind(id).fetch_one(&self.pool).await?;
Ok(user.into())
}
pub async fn delete_user(&self, id: i32) -> Result<(), ApiError> {
sqlx::query("DELETE FROM users WHERE id = $1").bind(id).execute(&self.pool).await?;
Ok(())
}
}
缓存管理使用 Redis 库，缓存用户信息，提高系统的响应速度。
用户缓存：
// src/cache/mod.rs
use crate::config::RedisConfig;
use crate::api::user::UserResponse;
use redis::AsyncCommands;
use serde_json;
use std::time::Duration;
pub struct UserCache {
client: redis::Client,
}
impl UserCache {
pub fn new(config: &RedisConfig) -> Self {
let client = redis::Client::open(&config.url).unwrap();
UserCache { client }
}
pub async fn get_user(&self, id: i32) -> Result<Option<UserResponse>, crate::error::ApiError> {
let mut con = self.client.get_tokio_connection().await?;
let cached = con.get::<_, Option<String>>(format!("user:{}", id)).await?;
if let Some(cached) = cached {
Ok(Some(serde_json::from_str(&cached)?))
} else {
Ok(None)
}
}
pub async fn put_user(&self, id: i32, user: &UserResponse) -> Result<(), crate::error::ApiError> {
let mut con = self.client.get_tokio_connection().await?;
let serialized = serde_json::to_string(user)?;
con.set_ex(format!("user:{}", id), serialized, Duration::from_secs(3600)).await?;
Ok(())
}
pub async fn remove_user(&self, id: i32) -> Result<(), crate::error::ApiError> {
let mut con = self.client.get_tokio_connection().await?;
con.del(format!("user:{}", id)).await?;
Ok(())
}
}
消息队列管理使用 RabbitMQ 库，实现异步消息处理，如发送邮件和通知。
邮件生产者：
// src/mq/mod.rs
use crate::config::RabbitMQConfig;
use lapin::{options::*, types::FieldTable, Connection, ConnectionProperties, BasicProperties};
use tokio_amqp::LapinTokioExt;
use serde_json;
#[derive(Debug, serde::Serialize, serde::Deserialize)]
pub struct EmailMessage {
pub to: String,
pub subject: String,
pub body: String,
}
pub struct EmailProducer {
channel: lapin::Channel,
queue_name: String,
}
impl EmailProducer {
pub async fn new(config: &RabbitMQConfig) -> Result<Self, crate::error::ApiError> {
let conn = Connection::connect(&config.uri, ConnectionProperties::default().with_tokio()).await?;
let channel = conn.create_channel().await?;
let queue = channel.queue_declare(&config.email_queue, QueueDeclareOptions {
durable: true,
auto_delete: false,
exclusive: false,
passive: false,
..Default::default()
}, FieldTable::default()).await?;
println!("Declared email queue: {:?}", queue);
Ok(EmailProducer { channel, queue_name: config.email_queue.clone() })
}
pub async fn send_email(&self, message: EmailMessage) -> Result<(), crate::error::ApiError> {
let msg = serde_json::to_vec(&message)?;
self.channel.basic_publish("", &self.queue_name, BasicPublishOptions::default(), msg.as_slice(), BasicProperties::default().with_delivery_mode(2))
.await?.await?;
println!("Sent email message: {:?}", message);
Ok(())
}
}
身份认证使用 JWT，通过自定义中间件实现。
JWT 身份认证中间件：
// src/middleware/auth.rs
use crate::config::JwtConfig;
use crate::error::ApiError;
use axum::extract::{FromRequestParts, State};
use axum::http::{request, Request, StatusCode};
use axum::middleware::Next;
use axum::response::Response;
use jsonwebtoken::{decode, encode, Header, Validation};
use serde::{Deserialize, Serialize};
use std::time::Duration;
use chrono::{DateTime, Utc};
#[derive(Debug, Serialize, Deserialize)]
pub struct Claims {
pub id: i32,
pub name: String,
pub email: String,
pub exp: i64,
pub iat: i64,
}
impl Claims {
pub fn new(id: i32, name: &str, email: &str, expiry: i64) -> Self {
let now = Utc::now();
let exp = now + chrono::Duration::seconds(expiry);
Claims {
id,
name: name.to_string(),
email: email.to_string(),
exp: exp.timestamp(),
iat: now.timestamp(),
}
}
}
pub struct JwtAuthMiddleware {
config: JwtConfig,
}
impl JwtAuthMiddleware {
pub fn new(config: JwtConfig) -> Self {
JwtAuthMiddleware { config }
}
pub async fn middleware<B>(State(config): State<JwtConfig>, req: Request<B>, next: Next<B>) -> Result<Response, ApiError> {
let headers = req.headers();
let token = headers.get("Authorization").and_then(|value| value.to_str().ok()).and_then(|s| s.strip_prefix("Bearer "));
if let Some(token) = token {
let validation = Validation::default();
let decoded = decode::<Claims>(token, config.secret.as_bytes(), &validation).map_err(|_| ApiError::Unauthorized)?;
let mut req = req;
req.extensions_mut().insert(decoded.claims);
Ok(next.run(req).await)
} else {
Err(ApiError::Unauthorized)
}
}
pub fn encode_claims(&self, claims: &Claims) -> Result<String, ApiError> {
let token = encode(&Header::default(), claims, self.config.secret.as_bytes()).map_err(|_| ApiError::InternalServerError("Failed to encode JWT".to_string()))?;
Ok(token)
}
}
用户 API 实现用户的 CRUD 操作，使用前面定义的业务逻辑、数据访问和缓存管理。
用户 API：
// src/api/user.rs
use axum::{extract::{Path, Query, Json, State, Extension}, routing::{get, post, put, delete}, Router, http::StatusCode};
use serde::Deserialize;
use serde::Serialize;
use chrono::prelude::*;
use crate::error::ApiError;
use crate::db::user_repo::UserRepository;
use crate::cache::UserCache;
use crate::mq::EmailProducer;
use crate::mq::EmailMessage;
#[derive(Debug, Deserialize)]
pub struct PageParams {
page: Option<u32>,
page_size: Option<u32>,
}
#[derive(Debug, Deserialize)]
pub struct CreateUserRequest {
name: String,
email: String,
phone: Option<String>,
}
#[derive(Debug, Deserialize)]
pub struct UpdateUserRequest {
name: String,
email: String,
phone: Option<String>,
}
#[derive(Debug, Serialize)]
pub struct UserResponse {
id: i32,
name: String,
email: String,
phone: Option<String>,
created_at: NaiveDateTime,
updated_at: NaiveDateTime,
}
async fn get_user(
Path(id): Path<i32>,
State(repo): State<UserRepository>,
State(cache): State<UserCache>,
) -> Result<Json<UserResponse>, ApiError> {
if let Some(user) = cache.get_user(id).await? {
Ok(Json(user))
} else if let Some(user) = repo.get_user(id).await? {
cache.put_user(id, &user).await?;
Ok(Json(user))
} else {
Err(ApiError::NotFound)
}
}
async fn get_users(
Query(params): Query<PageParams>,
State(repo): State<UserRepository>,
) -> Result<Json<Vec<UserResponse>>, ApiError> {
let page = params.page.unwrap_or(1);
let page_size = params.page_size.unwrap_or(10);
let users = repo.get_users(page, page_size).await?;
Ok(Json(users))
}
async fn create_user(
Json(req): Json<CreateUserRequest>,
State(repo): State<UserRepository>,
State(cache): State<UserCache>,
State(email_producer): State<EmailProducer>,
) -> Result<(StatusCode, Json<UserResponse>), ApiError> {
let user = repo.create_user(req).await?;
cache.put_user(user.id, &user).await?;
// 发送邮件通知
let email_message = EmailMessage {
to: user.email.clone(),
subject: "欢迎使用用户管理系统".to_string(),
body: format!("您好，{}！欢迎使用用户管理系统。", user.name),
};
email_producer.send_email(email_message).await?;
Ok((StatusCode::CREATED, Json(user)))
}
async fn update_user(
Path(id): Path<i32>,
Json(req): Json<UpdateUserRequest>,
State(repo): State<UserRepository>,
State(cache): State<UserCache>,
) -> Result<Json<UserResponse>, ApiError> {
let user = repo.update_user(id, req).await?;
cache.put_user(id, &user).await?;
Ok(Json(user))
}
async fn delete_user(
Path(id): Path<i32>,
State(repo): State<UserRepository>,
State(cache): State<UserCache>,
) -> Result<StatusCode, ApiError> {
repo.delete_user(id).await?;
cache.remove_user(id).await?;
Ok(StatusCode::NO_CONTENT)
}
pub fn create_user_routes(
repo: UserRepository,
cache: UserCache,
email_producer: EmailProducer,
) -> Router {
Router::new()
.route("/:id", get(get_user))
.route("", get(get_users))
.route("", post(create_user))
.route("/:id", put(update_user))
.route("/:id", delete(delete_user))
.with_state(repo)
.with_state(cache)
.with_state(email_producer)
}
身份认证 API 实现用户的登录和注册功能。
身份认证 API：
// src/api/auth.rs
use axum::{extract::{Json, State}, routing::{post}, Router, http::StatusCode};
use serde::Deserialize;
use serde::Serialize;
use crate::error::ApiError;
use crate::db::user_repo::UserRepository;
use crate::middleware::auth::{Claims, JwtAuthMiddleware};
#[derive(Debug, Deserialize)]
pub struct LoginRequest {
email: String,
password: String,
}
#[derive(Debug, Serialize)]
pub struct LoginResponse {
token: String,
}
#[derive(Debug, Deserialize)]
pub struct RegisterRequest {
name: String,
email: String,
phone: Option<String>,
password: String,
}
async fn login(
Json(req): Json<LoginRequest>,
State(repo): State<UserRepository>,
State(jwt_auth): State<JwtAuthMiddleware>,
) -> Result<Json<LoginResponse>, ApiError> {
// 查询用户
let users = repo.get_users(1, 10).await?;
let user = users.into_iter().find(|u| u.email == req.email);
if let Some(user) = user {
// 验证密码
let valid = validate_password(&req.password, &user.password_hash)?;
if valid {
// 生成 JWT token
let claims = Claims::new(user.id, &user.name, &user.email, 3600);
let token = jwt_auth.encode_claims(&claims)?;
Ok(Json(LoginResponse { token }))
} else {
Err(ApiError::Unauthorized)
}
} else {
Err(ApiError::NotFound)
}
}
async fn register(
Json(req): Json<RegisterRequest>,
State(repo): State<UserRepository>,
State(jwt_auth): State<JwtAuthMiddleware>,
) -> Result<(StatusCode, Json<LoginResponse>), ApiError> {
// 创建用户
let create_req = crate::api::user::CreateUserRequest {
name: req.name.clone(),
email: req.email.clone(),
phone: req.phone.clone(),
};
let user = repo.create_user(create_req).await?;
// 生成 JWT token
let claims = Claims::new(user.id, &user.name, &user.email, 3600);
let token = jwt_auth.encode_claims(&claims)?;
Ok((StatusCode::CREATED, Json(LoginResponse { token })))
}
fn validate_password(password: &str, password_hash: &str) -> Result<bool, ApiError> {
// 简单的密码验证，实际应该使用 bcrypt 或 Argon2
Ok(password == password_hash)
}
pub fn create_auth_routes(
repo: UserRepository,
jwt_auth: JwtAuthMiddleware,
) -> Router {
Router::new()
.route("/login", post(login))
.route("/register", post(register))
.with_state(repo)
.with_state(jwt_auth)
}
主函数初始化配置、数据库连接、缓存、消息队列和 API 路由。
主函数：
// src/main.rs
use user_management_api::config::AppConfig;
use user_management_api::db::user_repo::UserRepository;
use user_management_api::cache::UserCache;
use user_management_api::mq::EmailProducer;
use user_management_api::middleware::auth::JwtAuthMiddleware;
use user_management_api::api::user::create_user_routes;
use user_management_api::api::auth::create_auth_routes;
use user_management_api::middleware::cors::CorsMiddleware;
use user_management_api::middleware::compression::CompressionMiddleware;
use axum::{routing::{get}, Router};
use tracing_subscriber;
use tokio;
async fn health() -> impl axum::response::IntoResponse { "Healthy" }
#[tokio::main]
async fn main() {
// 初始化日志
tracing_subscriber::fmt::init();
// 加载配置
let config = AppConfig::from_env().expect("Failed to load config");
// 初始化数据库连接池
let user_repo = UserRepository::new(&config.database);
// 初始化 Redis 缓存
let user_cache = UserCache::new(&config.redis);
// 初始化 RabbitMQ 邮件生产者
let email_producer = EmailProducer::new(&config.rabbitmq).await.expect("Failed to connect to RabbitMQ");
// 初始化 JWT 身份认证
let jwt_auth = JwtAuthMiddleware::new(config.jwt.clone());
// 创建 API 路由
let user_routes = create_user_routes(user_repo.clone(), user_cache.clone(), email_producer.clone());
let auth_routes = create_auth_routes(user_repo.clone(), jwt_auth.clone());
let app = Router::new()
.route("/health", get(health))
.nest("/api/users", user_routes)
.nest("/api/auth", auth_routes)
.with_state(config)
.with_state(user_repo)
.with_state(user_cache)
.with_state(email_producer)
.with_state(jwt_auth)
.layer(axum::middleware::from_fn(JwtAuthMiddleware::middleware))
.layer(CorsMiddleware::layer())
.layer(CompressionMiddleware::layer());
// 启动服务器
let address = format!("{}:{}", config.server.address, config.server.port);
tracing::info!("Server listening on {}", address);
axum::Server::bind(&address.parse().expect("Invalid address"))
.serve(app.into_make_service())
.await
.expect("Failed to start server");
}
Axum 框架基于 Tokio 运行时，支持高并发的请求处理。可以通过调整 Tokio 的配置来提高并发性能。
调整 Tokio 配置：
use tokio::runtime::Builder;
#[tokio::main]
async fn main() {
let runtime = Builder::new_multi_thread()
.worker_threads(8) // 设置工作线程数
.max_blocking_threads(16) // 设置阻塞线程数
.enable_all()
.build()
.unwrap();
runtime.block_on(async {
// 应用初始化和启动代码
});
}
使用 Redis 缓存可以提高系统的响应速度。可以使用不同的缓存策略，如 TTL（Time To Live）和 LRU（Least Recently Used）。
缓存策略：
use std::time::Duration;
pub async fn put_user(&self, id: i32, user: &UserResponse) -> Result<(), crate::error::ApiError> {
let mut con = self.client.get_tokio_connection().await?;
let serialized = serde_json::to_string(user)?;
con.set_ex(format!("user:{}", id), serialized, Duration::from_secs(3600)).await?;
Ok(())
}
使用 SQLx 库访问数据库时，可以使用连接池和查询优化来提高数据库的性能。
连接池：
use sqlx::postgres::PgPoolOptions;
impl UserRepository {
pub fn new(config: &DatabaseConfig) -> Self {
let pool = PgPoolOptions::new()
.max_connections(10) // 设置连接池大小
.connect_lazy(&config.url)
.unwrap();
UserRepository { pool }
}
}
查询优化：
async fn get_user(&self, id: i32) -> Result<Option<UserResponse>, ApiError> {
let user = sqlx::query_as::<_, User>("SELECT id, name, email, phone, created_at, updated_at FROM users WHERE id = $1")
.bind(id)
.fetch_optional(&self.pool)
.await?;
Ok(user.map(|u| u.into()))
}
使用压缩和序列化优化可以减少网络传输和响应时间。Axum 框架支持 gzip、deflate 和 br 压缩。
压缩中间件：
use tower_http::compression::CompressionLayer;
let app = Router::new().layer(CompressionLayer::new());
序列化优化：
use serde_json::to_vec;
pub async fn get_user(&self, id: i32) -> Result<Option<UserResponse>, ApiError> {
let user = repo.get_user(id).await?;
Ok(user.map(|u| u.into()))
}
Axum 应用可以使用 Docker 容器化部署，或者直接在服务器上运行。
Docker 文件：
# Dockerfile
FROM rust:1.70-slim AS build
WORKDIR /app
COPY Cargo.toml Cargo.lock ./
RUN cargo init --name user-management-api --bin
COPY src/ src/
COPY static/ static/
RUN cargo build --release
FROM rust:1.70-slim AS runtime
WORKDIR /app
COPY --from=build /app/target/release/user-management-api /usr/local/bin/
COPY --from=build /app/static /app/static
EXPOSE 3000
CMD ["user-management-api"]
Docker Compose：
version:"3.8"
services:
app:
build: .
ports:
- "3000:3000"
environment:
- APP__DATABASE__URL=postgres://postgres:password@db:5432/postgres
- APP__REDIS__URL=redis://redis:6379
- APP__RABBITMQ__URI=amqp://guest:guest@rabbitmq:5672/%2F
- APP__JWT__SECRET=supersecretkey
- APP__JWT__EXPIRY=3600
- APP__SERVER__ADDRESS=0.0.0.0
- APP__SERVER__PORT=3000
depends_on:
- db
- redis
- rabbitmq
db:
image: postgres:15
ports:
- "5432:5432"
environment:
- POSTGRES_USER=postgres
- POSTGRES_PASSWORD=password
- POSTGRES_DB=postgres
redis:
image: redis:7
ports:
- "6379:6379"
rabbitmq:
image: rabbitmq:3-management
ports:
- "5672:5672"
- "15672:15672"
environment:
- RABBITMQ_DEFAULT_USER=guest
- RABBITMQ_DEFAULT_PASS=guest
使用 Tracing 和 Prometheus 等工具可以监控 Axum 应用的性能和状态。
Tracing：
use tracing_subscriber;
#[tokio::main]
async fn main() {
tracing_subscriber::fmt::init();
// 应用初始化和启动代码
}
Prometheus：
use prometheus_client::encoding::text::encode;
use prometheus_client::registry::Registry;
use axum::{routing::get, Router};
async fn metrics() -> impl axum::response::IntoResponse {
let mut registry = Registry::default();
// 注册指标
let response = encode(®istry).expect("Failed to encode metrics");
response
}
#[tokio::main]
async fn main() {
let app = Router::new().route("/metrics", get(metrics));
// 服务器启动代码
}
异步 API 设计是现代 Web 应用和微服务架构中的重要组成部分，通过非阻塞性的请求处理，显著提高了系统的并发能力和响应速度。Rust 语言的异步特性和 Axum 框架的简洁设计，使得构建高性能、可靠的异步 API 变得更加容易。
本章深入探讨了异步 API 的设计原则，介绍了 Axum 框架的核心功能，包括路由设计、中间件、错误处理、参数提取和响应构建，并通过实战项目集成演示了如何构建高性能的异步 API。我们还介绍了性能优化方法和部署与监控方法，帮助读者在实际项目中应用异步 API 设计。
微信公众号「极客日志」，在微信中扫描左侧二维码关注。展示文案：极客日志 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