关于Actix的“提取器”（Extractors）

目录

1. 概览 — Actix 的“提取器”（Extractors）
2. 1) Typed JSON（推荐，用 serde 定义结构体）
3. 2) 任意 JSON（serde_json::Value）
4. 3) Multipart 文件上传（multipart/form-data）
5. 4) application/x-www-form-urlencoded（web::Form<T>）
6. 5) 原始 body / 流式读取（大文件或自定义协议）
7. 6) 组合用法：同时读取 path / query / header / json
8. 7) 自定义提取器（实现 FromRequest）
9. 8) JSON size limit / 错误定制（JsonConfig）
10. 9) 错误类型与 ApiResult
11. 10) 实战建议与安全注意
12. 快速参考表

概览 — Actix 的“提取器”（Extractors）#

Actix 把很多常见请求部件做成了“提取器”，直接作为 handler 的参数就能拿到：

• web::Json<T>：解析 application/json 到 T: Deserialize（T 可以是 serde_json::Value）。
• web::Form<T>：解析 application/x-www-form-urlencoded。
• web::Path<T>：路径参数（/users/{id}）。
• web::Query<T>：查询字符串 ?page=1。
• web::Data<T>：应用级共享状态（例如 PgPool）。
• HttpRequest：可以读取 headers、连接信息等。
• web::Bytes / web::Payload：拿到原始 body 或流式读取。
• actix_multipart::Multipart：处理 multipart/form-data（文件上传）。
• 可实现 FromRequest 的自定义提取器（例如认证用户 AuthUser）。

1) Typed JSON（推荐，用 serde 定义结构体）#

优点：类型安全、自动反序列化、易验证。

use actix_web::{post, web, HttpResponse, Result};
use serde::Deserialize;

#[derive(Deserialize)]
struct SyncSourcesReq {
    // 举例字段
    sources: Vec<String>,
    dry_run: Option<bool>,
}

#[post("/sync/source")]
async fn sync_task_source(
    db_pool: web::Data<PgPool>,
    req: web::Json<SyncSourcesReq>,
) -> Result<HttpResponse> {
    let body: SyncSourcesReq = req.into_inner(); // 移出装箱数据
    // 使用 db_pool, body.sources ...
    Ok(HttpResponse::Ok().json(serde_json::json!({"ok": true})))
}

如果客户端没有 Content-Type: application/json，web::Json<T> 会返回错误（默认是 415/400 类错误）。可以通过 JsonConfig 自定义行为（见后）。

2) 任意 JSON（serde_json::Value）#

当你要接收任意结构（或不想先建 struct）：

use serde_json::Value;

#[post("/sync/source")]
async fn sync_task_source(
    db_pool: web::Data<PgPool>,
    web::Json(sources): web::Json<Value>, // 直接解构绑定 `sources`
) -> Result<HttpResponse> {
    // sources 是 serde_json::Value，可以用索引 / as_* 读取
    if let Some(array) = sources.get("sources").and_then(|v| v.as_array()) {
        // ...
    }
    Ok(HttpResponse::Ok().finish())
}

3) Multipart 文件上传（multipart/form-data）#

示例保存上传文件到磁盘（异步）：

use actix_multipart::Multipart;
use futures_util::StreamExt;
use tokio::io::AsyncWriteExt;

async fn upload(mut payload: Multipart) -> Result<HttpResponse, actix_web::Error> {
    while let Some(item) = payload.next().await {
        let mut field = item?; // Field
        let cd = field.content_disposition();
        let filename = cd
            .get_filename()
            .map(|n| sanitize_filename::sanitize(n))
            .unwrap_or_else(|| "file".into());
        let filepath = format!("./tmp/{}", filename);
        let mut f = tokio::fs::File::create(&filepath).await?;
        while let Some(chunk) = field.next().await {
            let data = chunk?;
            f.write_all(&data).await?;
        }
    }
    Ok(HttpResponse::Ok().body("uploaded"))
}

注意点：推荐使用非阻塞 tokio::fs。为 filename 使用 sanitize_filename 等库以防路径注入。

4) application/x-www-form-urlencoded（web::Form<T>）#

#[derive(Deserialize)]
struct MyForm { name: String, age: Option<u8> }

async fn handle_form(form: web::Form<MyForm>) -> Result<HttpResponse> {
    let f = form.into_inner();
    Ok(HttpResponse::Ok().json(f))
}

5) 原始 body / 流式读取（大文件或自定义协议）#

• 完全缓冲（body 较小时）：

async fn raw_body(body: web::Bytes) -> Result<HttpResponse> {
    // body: 全部字节（内存中）
    Ok(HttpResponse::Ok().body(body))
}

• 流式读取（需要限制大小并逐块处理）：

use futures_util::StreamExt;
use actix_web::web::BytesMut;

async fn stream_body(mut payload: web::Payload) -> Result<HttpResponse> {
    let mut body = BytesMut::new();
    while let Some(chunk) = payload.next().await {
        let data = chunk?;
        body.extend_from_slice(&data);
        if body.len() > 10_000_000 { // 例如 10MB 限制
            return Err(actix_web::error::ErrorPayloadTooLarge("payload too large"));
        }
    }
    // body.freeze() -> Bytes
    Ok(HttpResponse::Ok().body(body))
}

6) 组合用法：同时读取 path / query / header / json#

#[derive(Deserialize)]
struct PathInfo { id: i32 }

#[derive(Deserialize)]
struct Page { page: Option<u32> }

#[derive(Deserialize)]
struct CreateReq { title: String }

#[post("/users/{id}/posts")]
async fn create_post(
    db: web::Data<PgPool>,
    path: web::Path<PathInfo>,
    query: web::Query<Page>,
    req: actix_web::HttpRequest,
    body: web::Json<CreateReq>,
) -> Result<HttpResponse> {
    let id = path.id;
    let page = query.page.unwrap_or(1);
    let auth = req.headers().get("authorization").and_then(|h| h.to_str().ok());
    let body = body.into_inner();
    // ...
    Ok(HttpResponse::Created().finish())
}

7) 自定义提取器（实现 FromRequest）#

当很多 handler 都需要同样逻辑（例如从 header 解 token 并校验）时，写一个 AuthUser 提取器更整洁：

use actix_web::{FromRequest, HttpRequest, dev::Payload, Error};
use futures_util::future::{ready, Ready};

struct AuthUser { user_id: uuid::Uuid }

impl FromRequest for AuthUser {
    type Error = Error;
    type Future = Ready<Result<Self, Self::Error>>;
    fn from_request(req: &HttpRequest, _: &mut Payload) -> Self::Future {
        if let Some(h) = req.headers().get("authorization").and_then(|v| v.to_str().ok()) {
            // parse/validate token -> user_id
            // if ok:
            // return ready(Ok(AuthUser{ user_id }));
        }
        // otherwise
        ready(Err(actix_web::error::ErrorUnauthorized("no auth")))
    }
}

然后在 handler 中直接 auth: AuthUser。

8) JSON size limit / 错误定制（JsonConfig）#

全局或局部限制 JSON 大小并自定义错误响应：

use actix_web::{web, App, HttpResponse};

App::new()
    .app_data(web::JsonConfig::default()
        .limit(8192) // bytes
        .error_handler(|err, _req| {
            actix_web::error::InternalError::from_response(
                err,
                HttpResponse::BadRequest().json(serde_json::json!({"error": "invalid json"}))
            ).into()
        }))
    .service(sync_task_source);

9) 错误类型与 ApiResult#

你示例里 -> ApiResult {}，通常 ApiResult 是某种 Result<T, ApiError>。示例自定义错误并实现 ResponseError：

use actix_web::{ResponseError, HttpResponse};
use serde::Serialize;
use http::StatusCode;

#[derive(Debug)]
struct ApiError { msg: String, status: StatusCode }

impl ResponseError for ApiError {
    fn status_code(&self) -> StatusCode { self.status }
    fn error_response(&self) -> HttpResponse {
        HttpResponse::build(self.status).json(serde_json::json!({"error": self.msg}))
    }
}

type ApiResult<T = HttpResponse> = Result<T, ApiError>;

在 handler 中把 serde 错误或 DB 错误映射为 ApiError 即可。

10) 实战建议与安全注意#

• 限制 body 大小（防止 DoS）：使用 JsonConfig::limit、在流式读取时检查总大小。
• 校验与白名单字段：serde 的 #[serde(deny_unknown_fields)] 可拒绝多余字段（根据需要）。
• 不要进行阻塞 IO：用 tokio::fs 或 web::block 把阻塞操作推到线程池。
• 验证文件名与路径：对上传文件名做 sanitize，避免路径穿越。
• 检查 Content-Type：不要只信任扩展名，必要时手动检查 Content-Type。
• 捕获与转化解析错误：给客户端返回友好的错误消息和正确的 HTTP 状态码（400/415/413 等）。
• CORS / CSRF：按需配置中间件。
• 日志和追踪：用 tracing/log 记录 payload 大小、解析错误等。

快速参考表#

• JSON（typed）：web::Json<MyReq>
• JSON（动态）：web::Json<serde_json::Value> 或 Bytes/Payload 手动解析
• Form：web::Form<T>
• Query：web::Query<T>
• Path：web::Path<T> 或 web::Path<(u32, String)>
• Headers：通过 HttpRequest::headers()
• File upload：actix_multipart::Multipart
• 共享状态：web::Data<T>
• 自定义 auth：实现 FromRequest

---