Rust 中的中间件系统设计与实践指南

在现代服务架构中，“业务逻辑”只是请求旅程的一段。请求还要经过认证、限流、日志、指标、重试、缓存、故障恢复等关卡。这些横切关注点若直接灌入业务层，不但耦合度高、难以维护，还会导致“胶水代码”泛滥。中间件（Middleware）系统正是解决这个问题的关键。Rust 在类型系统、零成本抽象和所有权管理方面具备独特优势，使得中间件设计既能保持性能，又能保持语义清晰。本文从基础理念、执行模型、常见模式，到组合策略、性能调优和测试观测，深入解析如何在 Rust 中设计健壮的中间件系统，并结合 Actix、Axum（Tower）等生态提供实用示例。

1. 中间件的角色：把横切逻辑调度成管线

中间件位于“传输层到业务逻辑”的中间层，负责以可组合的方式处理横切关注点。典型职责包括：

• 认证授权：验证身份、角色，拒绝非法访问。
• 日志与追踪：记录请求、响应及上下游 trace 信息。
• 限流与熔断：保护后端资源免受突发流量冲击。
• 缓存与压缩：优化数据传输效率。
• 安全控制：注入安全头、防御攻击。
• 指标采集：暴露请求延迟、失败率等监控数据。

中间件链将这些职责以“洋葱模型”环绕 Handler。每个中间件接受请求、执行特定逻辑，再将请求移交链上的下一个中间件，直至最终 Handler 生成响应后，中间件链可以对响应进行修饰或记录。

Rust 最常见的中间件接口模式之一是 Service trait（来自 Tower），Actix、Axum、Tonic 等框架均基于此抽象构建。只要把握住“能理解不同框架的差异。

2. Service/Transform 模型：Tower 与 Actix 的共性

Service<Request> trait 定义了请求处理契约：

pub trait Service<Request> {
    type Response;
    type Error;
    type Future: Future<Output = Result<Self::Response, Self::Error>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>>;
    fn call(&mut self, req: Request) -> Self::Future;
}

• poll_ready：在发送请求前调用，允许中间件进行背压控制。
• call：接收请求并返回一个 future，其中包含异步响应流程。

中间件即一个包装了下游 service 的 Service。例如 logging 中间件：

use tower::{Service, ServiceBuilder};
use std::task::{Context, Poll};
use std::future::Future;
use std::pin::Pin;

#[derive(Clone)]
struct LoggingLayer;

impl<S> tower_layer::Layer<S> for LoggingLayer {
    type Service = LoggingMiddleware<S>;

    fn layer(&self, inner: S) -> Self::Service {
        LoggingMiddleware { inner }
    }
}

#[derive(Clone)]
struct LoggingMiddleware<S> {
    inner: S,
}

impl<S, Request> Service<Request> for LoggingMiddleware<S>
where
    S: Service<Request>,
    S::Future: Send + 'static,
    Request: std::fmt::Debug,
{
    type Response = S::Response;
    type Error = S::Error;
    type Future =
        Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, req: Request) -> Self::Future {
        println!("In***ing request: {:?}", req);
        let fut = self.inner.call(req);

        Box::pin(async move {
            let res = fut.await;
            println!("Response ready: {:?}", res.as_ref().map(|_| ()));
            res
        })
    }
}

• Layer 封装中间件构造逻辑；
• call 先打印请求，再调用下游 Service；
• 响应后打印日志。

在 Actix 中，中间件通过 Transform trait 实现，概念与 Tower layer 类似：Transform 接收 Service，返回包裹后的 Service。

use actix_web::dev::{Service, ServiceRequest, ServiceResponse, Transform};
use actix_web::{Error, HttpMessage};
use futures_util::future::{ok, Ready};

pub struct Logging;

impl<S, B> Transform<S, ServiceRequest> for Logging
where
    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error> + 'static,
{
    type Response = ServiceResponse<B>;
    type Error = Error;
    type Transform = LoggingMiddleware<S>;
    type InitError = ();
    type Future = Ready<Result<Self::Transform, Self::InitError>>;

    fn new_transform(&self, service: S) -> Self::Future {
        ok(LoggingMiddleware { service })
    }
}

pub struct LoggingMiddleware<S> {
    service: S,
}

impl<S, B> Service<ServiceRequest> for LoggingMiddleware<S>
where
    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error> + 'static,
{
    type Response = ServiceResponse<B>;
    type Error = Error;
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.service.poll_ready(cx)
    }

    fn call(&mut self, req: ServiceRequest) -> Self::Future {
        println!("request path: {}", req.path());
        let fut = self.service.call(req);
        Box::pin(async move {
            let res = fut.await?;
            println!("response status: {}", res.status());
            Ok(res)
        })
    }
}

无论 Tower 还是 Actix，模式类似：中间件 wrap 下游 service，并在 call 前后加入逻辑。

3. 常见中间件模式：从鉴权到限流

3.1 鉴权与认证

针对 HTTP 服务，我们常使用 Bearer token 或 session/cookie 识别用户。以 Axum/Tower 为例，可以编写一个提供 AuthContext 的 layer：

use axum::{
    async_trait,
    extract::{FromRequestParts},
    http::{Request, StatusCode},
    Router, routing::get,
    response::{IntoResponse}
};
use tower::{Layer, Service};
use futures_util::future::BoxFuture;

#[derive(Clone)]
struct AuthLayer;

impl<S> Layer<S> for AuthLayer {
    type Service = AuthMiddleware<S>;
    fn layer(&self, inner: S) -> Self::Service {
        AuthMiddleware { inner }
    }
}

#[derive(Clone)]
struct AuthMiddleware<S> {
    inner: S,
}

impl<S, B> Service<Request<B>> for AuthMiddleware<S>
where
    S: Service<Request<B>> + Clone + Send + 'static,
    S::Future: Send + 'static,
    B: Send + 'static,
{
    type Response = S::Response;
    type Error = S::Error;
    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;

    fn poll_ready(&mut self, cx: &mut std::task::Context<'_>) -> std::task::Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, mut req: Request<B>) -> Self::Future {
        let mut inner = self.inner.clone();
        Box::pin(async move {
            if let Some(auth_header) = req.headers().get("Authorization") {
                if is_valid(auth_header)? {
                    // inject auth data into extensions
                    req.extensions_mut().insert(AuthInfo::new(auth_header.clone()));
                    inner.call(req).await
                } else {
                    Err((
                        StatusCode::UNAUTHORIZED,
                        "Invalid token"
                    ).into())
                }
            } else {
                Err((
                    StatusCode::UNAUTHORIZED,
                    "Missing token"
                ).into())
            }
        })
    }
}

结合提取器 FromRequestParts 可在 handler 内获取 AuthInfo，实现认证/授权。

3.2 限流（Rate Limiting）

限流常用 Semaphore 或 Leaky Bucket 算法：

use std::{sync::Arc, time::Duration};
use tokio::sync::Semaphore;
use tower::{Layer, Service};

#[derive(Clone)]
struct RateLimitLayer {
    semaphore: Arc<Semaphore>,
}

impl RateLimitLayer {
    fn new(bound: usize) -> Self {
        Self { semaphore: Arc::new(Semaphore::new(bound)) }
    }
}

impl<S> Layer<S> for RateLimitLayer {
    type Service = RateLimitMiddleware<S>;
    fn layer(&self, inner: S) -> Self::Service {
        RateLimitMiddleware {
            inner,
            semaphore: self.semaphore.clone(),
        }
    }
}

#[derive(Clone)]
struct RateLimitMiddleware<S> {
    inner: S,
    semaphore: Arc<Semaphore>,
}

impl<S, Request> Service<Request> for RateLimitMiddleware<S>
where
    S: Service<Request>,
    S::Future: Send + 'static,
    Request: Send + 'static,
{
    type Response = S::Response;
    type Error = S::Error;
    type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;

    fn poll_ready(&mut self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.inner.poll_ready(cx)
    }

    fn call(&mut self, req: Request) -> Self::Future {
        let permit = self.semaphore.clone().acquire_owned();
        let mut inner = self.inner.clone();

        Box::pin(async move {
            let _guard = permit.await.expect("semaphore closed");
            inner.call(req).await
        })
    }
}

结合 timeout、wait queue 可实现更完整的限流/熔断策略。

3.3 Trace 与 Metrics

利用 tracing 或 opentelemetry 在中间件中记录链路：

use tracing::{info_span, Span};

fn main() {
    let app = Router::new()
        .route("/", get(handler))
        .layer(
            ServiceBuilder::new()
                .layer(TraceLayer::new_for_http())
                .layer(Extension(metrics_registry))
        );
}

TraceLayer::new_for_http() 自动启动/结束 span，捕获 request/response 数据。配合 Prometheus tower_http::metrics::InFlightRequests 可记录请求数量、响应时间。

3.4 重试与回退

借助 Tower 的 retry layer：

use tower::retry::Retry;
use tower::retry::policy::{Policy, RetryPolicy};

struct StatusRetry;

impl<E> Policy<Result<Response, E>, E> for StatusRetry {
    type Future = Ready<Self>;
    fn retry(&self, result: &Result<Response, E>, _error: Option<&E>) -> Option<Self::Future> {
        match result {
            Ok(res) if res.status().is_server_error() => Some(ready(Self)),
            Ok(_) | Err(_) => None,
        }
    }

    fn clone_request(&self, req: &Request<Body>) -> Option<Request<Body>> {
        req.try_clone()
    }
}

let client = tower::ServiceBuilder::new()
    .retry(StatusRetry)
    .service(base_client);

实战中需要注意 req.try_clone() 只对 small body/buffered body 生效，针对 streaming body 需设计幂等逻辑。

4. 组合策略：中间件链的构建原则

4.1 顺序与作用域

• 注册顺序决定执行顺序：早注册的 middleware 后执行；
• 典型顺序：Tracing -> Metrics -> Auth -> Timeout -> Handler；
• 在 Actix 中 wrap 按 App/Scope/Resource 层级生效；
• 组合 Tower layer 时可以用 ServiceBuilder 保持可读性。

例：

use tower::ServiceBuilder;

let middleware = ServiceBuilder::new()
    .layer(TraceLayer::new_for_http())
    .layer(***pressionLayer::new())
    .layer(TimeoutLayer::new(Duration::from_secs(2)))
    .layer(RateLimitLayer::new(100))
    .into_inner();

4.2 Shareable vs per-call 状态

• 无状态中间件（stateless）可以 clone（Tower layer 需要 Clone）；
• 有状态（如 per-tenant config）需要 Arc 包裹；
• Actix middleware 运行在 worker 内，可利用 Data<T> 注入共享状态。

4.3 错误处理

• 统一错误类型：实现 ResponseError 或 IntoResponse；
• 在 middleware 中拦截 .call() 结果并转换；
• 对 panic 使用 CatchPanic layer 保障安全。

5. 性能考量：中间件链不应成为瓶颈

5.1 同步 vs 异步

• 尽量避免在中间件 call 中执行阻塞操作；
• 若必须（例如读取本地文件），使用 tokio::task::spawn_blocking；
• Async 中间件内使用 Box::pin 会带来堆分配，Rust 允许使用 async fn + impl Future 规避（需要 async_trait 或手写 Pin）。

5.2 内存与分配

• Box::pin 频繁分配，可使用 pin_project 或 stack pinning；
• 对 heavy middleware（如 gzip 压缩）控制 buffer 大小；
• 共享 Arc/Semaphore 等资源避免 clone 大结构。

5.3 poll_ready 与背压

• poll_ready 允许中间件请求背压，例如 RateLimit；
• 在 poll_ready 中保持快速返回，不要执行阻塞；
• 合理使用 Semaphore，避免 deadlock（注意 Drop 顺序）。

6. 案例：构建一个多中间件 Web 服务

示例演示 Axum/Tower middleware 的组合，涵盖日志、指标、鉴权、限流、超时。

use axum::{routing::get, Router, Json, response::IntoResponse};
use serde::Serialize;
use tower::{ServiceBuilder, timeout::TimeoutLayer};
use tower_http::{
    trace::TraceLayer,
    classify::StatusInRangeAsFailures,
    ***pression::***pressionLayer,
    set_header::SetResponseHeaderLayer,
};

#[derive(Serialize)]
struct Health { status: &'static str }

async fn health_handler() -> impl IntoResponse {
    Json(Health { status: "ok" })
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    tracing_subscriber::fmt::init();

    let rate_limit = RateLimitLayer::new(100);     // 限流伪代码
    let auth_layer = AuthLayer;                    // 鉴权伪代码

    let middleware_stack = ServiceBuilder::new()
        .layer(TraceLayer::new_for_http())
        .layer(SetResponseHeaderLayer::if_not_present(
            http::header::CONTENT_TYPE,
            http::HeaderValue::from_static("application/json"),
        ))
        .layer(***pressionLayer::new())
        .layer(TimeoutLayer::new(Duration::from_secs(3)))
        .layer(rate_limit)
        .layer(auth_layer);

    let app = Router::new()
        .route("/health", get(health_handler))
        .layer(middleware_stack);

    axum::Server::bind(&"0.0.0.0:3000".parse().unwrap())
        .serve(app.into_make_service())
        .await?;

    Ok(())
}

说明：

• TraceLayer 输出 request/response 日志；
• SetResponseHeaderLayer 保证返回 JSON；
• ***pressionLayer 自动压缩响应；
• TimeoutLayer 防止 handler 超时；
• 自定义 RateLimitLayer, AuthLayer 完成限流与鉴权；
• ServiceBuilder 保证组合顺序可读；

此模式广泛用于API Gateway、微服务入口。

7. 中间件测试与验证

7.1 单元测试

Actix:

#[actix_rt::test]
async fn test_logging_mw() {
    let app = test::init_service(
        App::new()
            .wrap(Logging)
            .route("/", web::get().to(|| async { "hi" }))
    ).await;

    let req = test::TestRequest::get().uri("/").to_request();
    let resp = test::call_service(&app, req).await;
    assert!(resp.status().is_su***ess());
}

Axum/Tower:

use tower::ServiceExt; // for `oneshot`
#[tokio::test]
async fn test_auth_middleware() {
    let app = Router::new()
        .route("/", get(|| async { "OK" }))
        .layer(AuthLayer);

    let request = Request::builder().uri("/").body(Body::empty()).unwrap();
    let response = app.clone().oneshot(request).await.unwrap();
    assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
}

7.2 Integration Testing

结合 hyper::Client 或 reqwest 对 HTTP 服务进行集成测试。中间件结果可通过 header/response body/logging 观察。

8. 可观测性：把中间件链纳入监控体系

• Tracing：在每个 middleware 创建 span，attach request ID；
• Metrics：使用 Prometheus 或 OpenTelemetry，统计请求延迟、流量；
• 日志：中间件是天然的 log hook，建议在 ingress/egress 打点；
• 健康检查：中间件中注入 X-Request-Id，便于追踪；
• Debug 探针：提供 /metrics、/debug/vars、/status 等特殊 route。

在生产环境，必须确保中间件不会成为瓶颈或单点：监控 poll_ready 时间，观察 Semaphore 的 wait duration。

9. 总结与设计原则

1. 基于 Service/Layer 构建：遵循 Tower 抽象，保持中间件链可组合；
2. 明确执行顺序：对认证、限流、日志等中间件安排恰当顺序；
3. 区分 stateless/stateful：将共享资源封装于 Arc 或 Data；
4. 避免阻塞：中间件内部避免同步 I/O，必要时使用 spawn_blocking；
5. 善用 poll_ready：可实现背压、限流、Circuit Breaker；
6. 统一错误处理：使用 ResponseError/IntoResponse；
7. 可测试、可观察：封装中间件的同时提供测试与监控策略；
8. 性能意识：关注 Box::pin、Semaphore、timeout 开销；
9. 扩展性：中间件要支持 configuration 与多服务共享；
10. 文档化：将中间件链的顺序、逻辑写入设计文档，方便维护。

中间件系统是现代服务架构的“控场者”，它把横切关注点与业务逻辑隔离，构建可扩展、安全的服务。Rust 的强类型与 Service 抽象使得中间件设计既安全又高效；Actix、Tower/Axum、Tonic 等生态对这些模式做了充分探索。