一个上传图片的接口引发的思考

在我的博客开发中，后台提供了一个图片上传接口，给文章编辑器用。涉及到了一个复杂链路：

管理端调用后台的上传接口

后台再转发到资源服务的 /api/upload

资源服务在真正接收文件前，会回调后台校验上传 token

后台此时还在等待资源服务返回

这就形成了一个问题：我在等你，你在等我先处理完回调。

如果后台把这个上传请求卡在主事件循环里，资源服务的回调就进不来，整个链路会表现成超时、挂起。

链路里的关键代码

上传入口在后台路由里，原始处理方式是同步函数包一层异步接口：

@router.post("/upload", response_model=ApiResponse[MediaAssetUploadPayload])
async def upload_post_image(
    file: UploadFile = File(...),
    context: AdminAuthContext = Depends(require_permission("post:write")),
) -> ApiResponse[MediaAssetUploadPayload]:
    filename = file.filename or "upload-image"
    contents = await file.read()

    try:
        asset = await asyncio.to_thread(
            upload_media_asset,
            filename=filename,
            data=contents,
            auth_token=context.token,
        )
    except MediaAssetError as error:
        raise HTTPException(status_code=400, detail=str(error)) from error

资源服务地址和回调地址的组装在另一个地方：

def get_resource_upload_base_url() -> str:
    if settings.resource_upload_base_url and settings.resource_upload_base_url.strip():
        return settings.resource_upload_base_url.strip()

    raise MediaAssetError("RESOURCE_UPLOAD_BASE_URL is required")


def get_upload_token_validation_url() -> str:
    if settings.admin_api_base_url and settings.admin_api_base_url.strip():
        return f"{settings.admin_api_base_url.strip().rstrip('/')}/api/auth/validate-upload-token"

    raise MediaAssetError("ADMIN_API_BASE_URL is required")

真正上传时，会把 server_target 一并发给资源服务：

with httpx.Client(timeout=60) as client:
    response = client.post(
        f"{base_url}/api/upload",
        headers={"Authorization": f"Bearer {auth_token}"},
        files={"file": (filename, data, mime_type)},
        data={"server_target": server_target},
    )

资源服务收到这个 server_target 后，会回调后台的 token 校验接口：

@router.post("/validate-upload-token")
def validate_upload_token(
    authorization: str | None = Header(default=None),
    db: Session = Depends(get_db),
) -> dict[str, bool]:
    token = _parse_bearer_token(authorization)
    if not token:
        return {"valid": False}

    session = get_admin_session_by_token(db, token)
    if not session or not is_admin_session_active(session):
        return {"valid": False}

    user = session.user
    if not user.is_active:
        return {"valid": False}

    return {"valid": user_has_permission(db, user.id, "post:write")}

到这里，闭环已经非常清楚了。

如何处理

这个问题的处理方式其实不花哨，核心只有一句话：

不要让阻塞型同步 I/O 站在异步事件循环上。

我们没有一口气把整个上传链路重写成全异步，而是用了一个更实用的过渡方案：把同步上传逻辑丢到线程池里。

处理动作

把上传接口改成：

asset = await asyncio.to_thread(
    upload_media_asset,
    filename=filename,
    data=contents,
    auth_token=context.token,
)

这个改动的效果很直接：

upload_media_asset(...) 仍然保持同步实现

但它不再占着 FastAPI 的主事件循环

资源服务的回调请求能够正常打到后台

整条上传链路恢复可调度状态

为什么这招管用

因为问题的本质不是“上传慢”，而是“上传过程阻塞了调度器”。

asyncio.to_thread() 的价值就在于把这个阻塞点从事件循环里搬走，让异步框架继续处理别的请求，包括那个关键的回调验证请求。

事后复盘

这次事故最值得记住的，不是某个 API 名称，而是几个非常底层、但很容易在工程里踩坑的知识点。

`async` 不等于非阻塞

这是很多人第一次接触 Python 异步时最容易误解的点。

一个函数写成 async def，只代表它可以被事件循环调度，不代表函数体里做的事情天然就是异步的。

如果你在 async 函数内部继续执行同步 I/O，比如：

response = client.post(...)

那它照样会阻塞当前执行流。

事件循环是异步程序的“调度核心”

FastAPI / asyncio 这套模型的核心，不是“同时执行很多代码”，而是“在等待 I/O 的间隙把 CPU 让给别的任务”。

一旦你在事件循环里塞进一个长时间的同步请求，整个调度模型就会被破坏。不是程序崩了，而是它开始“安静地卡住”。

同步 I/O 和异步 I/O 的边界要非常敏感

这次上传链路里，真正危险的不是“发了 HTTP 请求”，而是“发 HTTP 请求时，它还需要别的服务回调我”。

这类调用一旦互相依赖，就很容易演化成：

A 等 B

B 等 A

最终大家一起等

`asyncio.to_thread()` 是很实用的工程折中

它不是最终形态，但在很多存量项目里非常好用。

适合场景：

旧逻辑已经写成同步函数，不想大改

但入口已经是异步 Web 框架

同步函数里又包含网络请求、文件读写等阻塞操作

它本质上是在告诉事件循环：这件事我去线程池里干，你继续调度别的请求。

回调闭环是服务设计里的高危点

任何“请求我方服务前，还要回调我方接口确认权限/状态”的设计，都要特别小心。

它天然增加了耦合，而且对并发模型很敏感。

如果主请求本身已经占住了处理能力，再来一个回调，系统就会表现得像自己在等自己。

其它

实际上这个链路一开始就不太符合现代化设计理念，更推荐使用如JWT做一次性签名和校验，但是由于考虑到为了这一个接口需要重新接入一个系统，时间成本上不太划算就没有采用。如果后续上传系统需要服务多个项目再进行改造。