Python 基础体系 · 第 81/112 篇。示例统一以 Python 3.14 为语言基线；第三方库使用与其兼容的现代稳定版本，版本敏感行为会单独说明。

Python WSGI 与 ASGI：调用协议、并发模型、中间件和部署选择

WSGI 和 ASGI 都不是 Web 框架，而是服务器与 Python 应用之间的调用协议。Flask、Django、FastAPI 等框架负责路由、请求解析、响应生成和业务组织；WSGI 或 ASGI 则规定服务器应如何调用框架，以及框架如何把结果交还给服务器。

理解二者的关键，不是先背诵“WSGI 是同步、ASGI 是异步”，而是回答四个问题：

1. 服务器调用应用时传入了什么？
2. 应用如何读取请求、写出响应？
3. 一次调用对应一个请求，还是一个长期连接？
4. 等待 I/O 时，其他请求是否能够继续推进？

WSGI 将 Web 抽象成一次同步的请求—响应调用；ASGI 将连接抽象成 scope 和一组异步事件。ASGI 3.0 规范明确规定，ASGI 应用是一个异步可调用对象，服务器每个连接调用一次应用，并通过 receive 和 send 交换事件。(asgi.readthedocs.io)

---

一、先区分三层：协议、框架和服务器

一个典型的 Python Web 请求链路可以表示为：

flowchart LR
    C[客户端] --> P[反向代理或负载均衡]
    P --> S[应用服务器]
    S --> A[WSGI 或 ASGI 应用]
    A --> F[Web 框架]
    F --> B[业务代码]
    B --> D[数据库、缓存、外部服务]

这里有三个容易混淆的角色：

• 协议：WSGI 或 ASGI，规定应用接口；
• 应用服务器：例如 Gunicorn、Uvicorn、Daphne、Hypercorn，负责监听网络、解析协议并调用应用；
• Web 框架：例如 Flask、Django、FastAPI，负责把底层请求转换成开发者熟悉的 request、路由参数和响应对象。

WSGI 规范本身并不规定服务器如何选择应用，也不规定线程数、进程数或配置方式；它只规定服务器和应用之间的公共接口。(peps.python.org) ASGI 也采用类似分层方式：协议服务器负责终止网络连接并转换为事件，应用负责处理事件并发送响应事件。(asgi.readthedocs.io)

因此：

Flask != WSGI
FastAPI != ASGI
Gunicorn != WSGI
Uvicorn != ASGI

更准确的关系是：

Flask 应用 --通常通过 WSGI--> WSGI 服务器
FastAPI 应用 --通过 ASGI--> ASGI 服务器

---

二、WSGI 的调用协议：一次调用，一次响应

2.1 WSGI 应用的最小形态

WSGI 应用必须接受两个位置参数：

def application(environ, start_response):
    ...

其中：

• environ 是包含请求信息的普通字典；
• start_response 是由服务器提供的回调，用于设置状态码和响应头；
• 应用最终返回一个可迭代对象，迭代得到响应体字节串。

下面是一个不依赖第三方库的完整 WSGI 应用：

# wsgi_app.py
from urllib.parse import parse_qs


def application(environ, start_response):
    method = environ["REQUEST_METHOD"]
    path = environ["PATH_INFO"]
    query = parse_qs(environ.get("QUERY_STRING", ""))

    if method == "GET" and path == "/":
        name = query.get("name", ["world"])[0]
        body = f"Hello, {name}!\n".encode("utf-8")

        start_response(
            "200 OK",
            [
                ("Content-Type", "text/plain; charset=utf-8"),
                ("Content-Length", str(len(body))),
            ],
        )
        return [body]

    body = b"Not Found\n"
    start_response(
        "404 Not Found",
        [
            ("Content-Type", "text/plain; charset=utf-8"),
            ("Content-Length", str(len(body))),
        ],
    )
    return [body]

请求：

GET /?name=Alice

服务器大致执行：

result = application(environ, start_response)

for chunk in result:
    send_to_client(chunk)

if hasattr(result, "close"):
    result.close()

environ 中常见字段包括：

{
    "REQUEST_METHOD": "GET",
    "PATH_INFO": "/",
    "QUERY_STRING": "name=Alice",
    "SERVER_NAME": "127.0.0.1",
    "SERVER_PORT": "8000",
    "wsgi.url_scheme": "http",
    "wsgi.input": request_body_stream,
}

WSGI 规范要求应用返回的可迭代对象产生字节串；应用也可以返回生成器，因此能够进行一定程度的流式响应。start_response 必须在第一个响应体字节产生之前被调用。(peps.python.org)

2.2 start_response 为什么是回调

WSGI 将响应拆成两部分：

start_response(status, headers)
返回的 iterable -> body bytes

例如：

def application(environ, start_response):
    start_response(
        "200 OK",
        [("Content-Type", "text/plain")],
    )

    yield b"first chunk\n"
    yield b"second chunk\n"

服务器先通过 start_response 获得响应元数据，再逐个消费响应体。

这是一种同步的拉取模型：

服务器调用应用
    应用返回 iterable
        服务器 next(iterable)
        应用产生一个 chunk
        服务器发送 chunk
        服务器继续 next(iterable)

应用不能在同一次调用中独立地等待多个未来事件，也没有规范化的异步 receive 和 send 通道。它适合传统 HTTP 请求—响应，但不适合 WebSocket 这类双向、长生命周期协议。ASGI 文档指出，即使把 WSGI 的单一调用改成异步函数，也仍然缺少接收多个 WebSocket 帧所需的事件通道。(asgi.readthedocs.io)

---

三、ASGI 的调用协议：连接范围加事件通道

3.1 ASGI 三参数接口

ASGI 3.0 应用的形态是：

async def application(scope, receive, send):
    ...

三个参数分别表示：

• scope：连接或请求的静态信息；
• receive：异步接收事件的可调用对象；
• send：异步发送事件的可调用对象。

ASGI 规范中的最小结构是：

async def application(scope, receive, send):
    event = await receive()
    ...
    await send({...})

scope 和事件都是字典，事件通过顶层的 type 字段区分类型。ASGI 3.0 的应用调用方式是单个异步可调用对象；早期 ASGI 2.0 的“两阶段调用”仍可能存在于旧应用中，但已经属于 legacy 形式。(asgi.readthedocs.io)

3.2 一个可运行的 ASGI HTTP 应用

# asgi_app.py
async def application(scope, receive, send):
    if scope["type"] != "http":
        raise RuntimeError(f"Unsupported protocol: {scope['type']}")

    request = await receive()

    if request["type"] != "http.request":
        raise RuntimeError(f"Unexpected event: {request['type']}")

    body = b"Hello from ASGI\n"

    await send(
        {
            "type": "http.response.start",
            "status": 200,
            "headers": [
                [b"content-type", b"text/plain; charset=utf-8"],
                [b"content-length", str(len(body)).encode("ascii")],
            ],
        }
    )

    await send(
        {
            "type": "http.response.body",
            "body": body,
            "more_body": False,
        }
    )

使用 Uvicorn 运行：

python -m pip install uvicorn
uvicorn asgi_app:application --host 127.0.0.1 --port 8000

其中：

asgi_app:application
│         │
│         └── 模块中的应用对象
└── Python 模块名

访问 http://127.0.0.1:8000/ 后，服务器会把 HTTP 请求转换为类似下面的事件：

{
    "type": "http.request",
    "body": b"",
    "more_body": False,
}

应用则发送两个事件：

{
    "type": "http.response.start",
    "status": 200,
    "headers": [...],
}

以及：

{
    "type": "http.response.body",
    "body": b"Hello from ASGI\n",
    "more_body": False,
}

ASGI HTTP 规范规定，HTTP 请求体可以通过多个 http.request 事件分块传递；响应也可以通过多个 http.response.body 事件分块发送。more_body=False 表示当前方向的数据已经结束。(asgi.readthedocs.io)

3.3 scope 与请求体的边界

ASGI 不把所有请求数据都塞进 scope。

例如 HTTP 的 scope 通常包含：

{
    "type": "http",
    "http_version": "1.1",
    "method": "POST",
    "scheme": "http",
    "path": "/upload",
    "query_string": b"chunked=true",
    "headers": [
        [b"host", b"localhost"],
        [b"content-type", b"application/octet-stream"],
    ],
}

但请求体通过 receive() 流入：

chunks = []

while True:
    event = await receive()

    if event["type"] == "http.disconnect":
        raise RuntimeError("Client disconnected")

    if event["type"] != "http.request":
        raise RuntimeError(f"Unexpected event: {event['type']}")

    chunks.append(event.get("body", b""))

    if not event.get("more_body", False):
        break

body = b"".join(chunks)

这里的因果关系是：

scope 保存“请求是什么”
receive() 提供“请求体正在到达”
more_body 决定“是否还要继续读取”

如果应用只调用一次 receive()，就可能只读到请求体的第一块。对于小 JSON 请求，服务器常常一次就提供完整数据，因此这个错误不容易暴露；对于文件上传、流式请求或大请求体，它会表现为数据截断。

---

四、WSGI 与 ASGI 的核心差异

维度	WSGI	ASGI
应用签名	application(environ, start_response)	async application(scope, receive, send)
调用方式	同步调用	异步调用并等待
请求模型	一个调用对应一个 HTTP 请求	一个调用对应一个协议连接或请求
输入	environ 与 wsgi.input	scope 与 receive() 事件
输出	返回可迭代的字节串	send() 发送事件
长连接	可做有限的 HTTP 流式响应	原生支持长轮询、WebSocket 等事件型协议
并发基础	进程、线程或服务器扩展	事件循环、协程、线程、进程
中间件	包装应用调用和响应 iterable	包装 scope、receive、send
典型框架	Flask、传统 Django	FastAPI、Starlette、Django Channels

需要特别注意：WSGI 不等于只能使用一个线程，ASGI 也不等于自动并行。WSGI 服务器可以使用多个进程或线程；ASGI 服务器也可以配置多个进程，并且 ASGI 应用内部仍然可以调用同步代码。

二者真正不同的是单次应用调用的交互能力：

WSGI：调用 -> 返回响应 iterable
ASGI：调用 -> receive/send 多次交互

---

五、并发模型：等待期间能否运行其他任务

5.1 并发不是并行

• 并发：多个任务在时间上交错推进；
• 并行：多个任务在同一时刻使用不同 CPU 核心执行。

在单线程事件循环中，ASGI 可以实现高效并发，但这不是 CPU 并行：

任务 A：发起数据库 I/O，等待
任务 B：继续执行
任务 B：发起网络 I/O，等待
任务 A：I/O 完成，恢复执行

Python 的 asyncio 提供了基于 async/await 的并发 I/O 基础；ASGI 应用通常运行在这样的事件循环之上。ASGI 规范要求应用是与 async/await 兼容的协程，但同步代码仍可放到线程或其他进程执行。(docs.python.org)

5.2 WSGI 的典型并发推导

假设有 100 个请求，每个请求都需要等待外部服务 1 秒，且服务器使用 4 个工作线程。忽略 CPU 和网络开销时：

每个线程一次处理一个请求
4 个线程同时处理 4 个请求
100 / 4 = 25 批
总等待时间约为 25 秒

如果线程数提高到 100，则理论等待时间接近 1 秒，但代价是：

• 更多线程栈和调度开销；
• 更多数据库连接；
• 同步锁竞争；
• 单进程资源上限更快耗尽。

5.3 ASGI 的典型并发推导

如果一个请求写成：

async def handler():
    result = await call_external_service()
    return result

当 call_external_service() 真正执行异步 I/O 时，协程在 await 处挂起，事件循环可以运行其他请求。

假设单个事件循环同时管理 100 个等待外部服务的协程：

请求 1 -> await 外部服务
请求 2 -> await 外部服务
...
请求 100 -> await 外部服务

外部服务陆续完成
协程分别恢复

这降低的是等待型并发的线程成本，不是把 100 个 CPU 密集型函数同时运行在一个 CPU 上。

5.4 ASGI 中最危险的阻塞代码

下面的代码虽然语法正确，但会阻塞事件循环：

import time


async def handler():
    time.sleep(5)
    return {"ok": True}

async def 只说明函数返回协程；它不会把同步函数自动变成非阻塞函数。执行 time.sleep(5) 时，事件循环线程被占用，其他协程也无法获得执行机会。

改成异步等待：

import asyncio


async def handler():
    await asyncio.sleep(5)
    return {"ok": True}

如果第三方库只有同步 API，应将其移入线程：

import asyncio


def blocking_call() -> dict:
    # 假设这是同步 SDK 调用
    return {"value": 42}


async def handler():
    result = await asyncio.to_thread(blocking_call)
    return result

但线程迁移不是免费操作。同步库可能：

• 持有线程局部状态；
• 依赖线程安全连接池；
• 在内部进行大量 CPU 计算；
• 对取消不敏感。

因此，asyncio.to_thread() 适合隔离阻塞 I/O，不应被误认为通用的 CPU 并行方案。

---

六、事件循环、线程和进程如何共同工作

一个常见的 ASGI 部署结构如下：

一个进程
└── 一个事件循环线程
    ├── 协程请求 A
    ├── 协程请求 B
    ├── 协程请求 C
    └── 线程池中的同步任务

多个工作进程则是：

主进程
├── Worker 1 -> Event Loop 1
├── Worker 2 -> Event Loop 2
├── Worker 3 -> Event Loop 3
└── Worker 4 -> Event Loop 4

进程之间不共享普通 Python 全局变量。因此下面的计数器不是整个服务的全局计数：

counter = 0

async def handler():
    global counter
    counter += 1
    return {"counter": counter}

如果部署了 4 个进程，每个进程都有自己的 counter。请求被负载均衡到不同进程后，客户端可能看到：

1, 2, 1, 3, 2, 1, ...

这不是 ASGI 的错误，而是进程隔离的必然结果。需要跨请求、跨线程、跨进程共享的状态，应放在数据库、Redis 等外部存储中，或者明确设计进程内缓存的一致性边界。

---

七、中间件：为什么 WSGI 和 ASGI 的写法不同

7.1 WSGI 中间件包装应用调用

WSGI 中间件通常是一个可调用对象：

from time import perf_counter


class TimingMiddleware:
    def __init__(self, app):
        self.app = app

    def __call__(self, environ, start_response):
        started = perf_counter()

        def timing_start_response(status, headers, exc_info=None):
            duration = perf_counter() - started
            headers = list(headers)
            headers.append(("X-Process-Time", f"{duration:.6f}"))
            return start_response(status, headers, exc_info)

        return self.app(environ, timing_start_response)

使用：

application = TimingMiddleware(application)

数据流是：

服务器
  -> TimingMiddleware.__call__
      -> inner_app(environ, wrapped_start_response)
          -> 返回 response iterable
      -> 中间件返回或包装 iterable

如果中间件要修改响应体，就必须包装返回的 iterable：

class PrefixMiddleware:
    def __init__(self, app):
        self.app = app

    def __call__(self, environ, start_response):
        result = self.app(environ, start_response)

        def generate():
            yield b"[prefix]\n"
            yield from result

        return generate()

生产实现还应处理 result.close()，否则下游生成器可能无法释放资源。WSGI 规范要求：如果返回的可迭代对象具有 close() 方法，服务器或中间件应在请求结束时调用它。(peps.python.org)

7.2 ASGI 中间件包装三个通道

ASGI 中间件本身也表现为 ASGI 应用：

class AddHeaderMiddleware:
    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        async def wrapped_send(message):
            if message["type"] == "http.response.start":
                message = dict(message)
                headers = list(message.get("headers", []))
                headers.append([b"x-service", b"demo"])
                message["headers"] = headers

            await send(message)

        await self.app(scope, receive, wrapped_send)

这里中间件没有直接返回响应，而是替换了 send：

服务器 receive/send
       |
       v
中间件 scope/receive/wrapped_send
       |
       v
内部应用 scope/receive/send

如果中间件要修改请求，可以包装 receive：

async def wrapped_receive():
    message = await receive()

    if message["type"] == "http.request":
        message = dict(message)
        # 修改前必须明确理解 more_body 和 body 的语义

    return message

如果中间件要增加 scope 信息，应先复制字典：

child_scope = dict(scope)
child_scope["myapp.user_id"] = "u-123"

await self.app(child_scope, receive, send)

ASGI 规范特别提醒，中间件修改 scope 时应先复制，因为直接修改可能影响上游中间件；并且应用收到的 scope 不一定就是中间件最初传下去的那个对象。(asgi.readthedocs.io)

7.3 中间件顺序会改变行为

假设有两个中间件：

A(B(application))

请求方向：

服务器 -> A -> B -> application

响应方向：

application -> B -> A -> 服务器

因此：

• 认证中间件放在业务中间件外层，业务中间件才能看到认证失败；
• 异常处理中间件必须位于可能抛错的组件外层；
• 压缩中间件必须理解响应是否分块；
• 日志中间件不能假设一个 ASGI 响应只有一个 body 事件。

把 WSGI 中间件直接挂到 ASGI 应用上通常会失败，因为两者调用签名不同：

# WSGI 期待两个参数
app(environ, start_response)

# ASGI 需要三个参数，并且返回协程
await app(scope, receive, send)

如果必须复用旧 WSGI 组件，应使用明确的适配器，而不是依赖参数数量“碰巧匹配”。

---

八、WSGI 与 ASGI 的互操作

ASGI 的 HTTP 子规范设计为 WSGI 格式的超集，并规定了可转换的 HTTP 请求集合；asgiref 提供了将 WSGI 应用放入 ASGI 服务器的兼容实现，通常会在线程池中运行同步 WSGI 应用，避免直接阻塞异步事件循环。(asgi.readthedocs.io)

概念上的适配过程是：

ASGI scope + receive()
        |
        v
转换为 WSGI environ + wsgi.input
        |
        v
在线程池调用 WSGI application()
        |
        v
收集或转发 WSGI iterable
        |
        v
转换为 ASGI response.start/body 事件

这类适配有清晰边界：

1. WSGI 应用只能看到 HTTP 请求；
2. WSGI 应用不能因此获得原生 WebSocket 能力；
3. 同步 WSGI 代码仍然占用线程；
4. 大响应体的转发需要考虑缓冲和反压；
5. 线程池容量会成为并发上限。

反方向的 ASGI 到 WSGI 适配更受限制。WSGI 没有原生的异步事件通道，也没有 WebSocket 语义，因此通常只能把有限的 HTTP 请求—响应场景转换过去。能够被转换，不代表两个协议能力等价。

---

九、FastAPI 为什么属于 ASGI 应用

FastAPI 是构建在 Starlette 之上的 ASGI Web 框架，同时使用 Python 类型标注完成参数解析、校验和 OpenAPI 文档生成。官方文档将 FastAPI 定义为面向 API 的现代 Python Web 框架，并明确列出 HTTP 与 WebSocket 支持。(fastapi.tiangolo.com)

一个完整的 FastAPI 示例：

# main.py
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request
from pydantic import BaseModel


class Item(BaseModel):
    name: str
    price: float


@asynccontextmanager
async def lifespan(app: FastAPI):
    print("startup: initialize resources")
    app.state.cache = {"ready": True}

    yield

    print("shutdown: release resources")
    app.state.cache.clear()


app = FastAPI(lifespan=lifespan)


@app.middleware("http")
async def add_process_header(request: Request, call_next):
    response = await call_next(request)
    response.headers["X-Service"] = "demo"
    return response


@app.get("/health")
async def health():
    return {"ok": True}


@app.post("/items")
async def create_item(item: Item):
    return {
        "name": item.name,
        "price": item.price,
    }

运行：

python -m pip install "fastapi[standard]"
fastapi dev

开发服务器启动后，可以访问：

http://127.0.0.1:8000/health
http://127.0.0.1:8000/docs

FastAPI 的同步路径函数也可以存在：

@app.get("/sync")
def sync_endpoint():
    return {"mode": "sync"}

异步路径函数则使用：

@app.get("/async")
async def async_endpoint():
    return {"mode": "async"}

重要的不是函数名，而是函数内部是否执行异步操作：

@app.get("/bad")
async def bad():
    import time
    time.sleep(2)  # 阻塞事件循环
    return {"ok": True}

这仍然是 ASGI 应用，但处理方式错误。FastAPI 不会把任意阻塞调用自动变成异步 I/O。相反，如果使用同步依赖或同步路径函数，框架通常会通过线程池隔离一部分同步执行；这属于框架实现策略，不应推导为“所有同步代码都没有成本”。

---

十、生命周期：请求生命周期和应用生命周期不是一回事

ASGI 的 lifespan 协议用于应用启动和关闭：

服务器准备启动
    -> lifespan.startup
应用初始化资源
    -> lifespan.startup.complete
服务器开始接受请求
    ...
服务器准备关闭
    -> lifespan.shutdown
应用释放资源
    -> lifespan.shutdown.complete

服务器必须等待 lifespan.startup.complete 后再处理连接；如果应用报告 lifespan.startup.failed，服务器应记录错误并退出。(asgi.readthedocs.io)

这适合初始化：

• 数据库连接池；
• HTTP 客户端；
• 模型；
• 缓存；
• 消息队列连接。

不适合把请求级数据放入全局资源。应区分：

应用级资源：进程启动时创建，进程关闭时释放
请求级资源：每个请求创建或借用，请求结束时释放

多进程部署时，生命周期会在每个 worker 进程中分别执行。因此初始化一个内存模型，可能会得到每个进程一份模型；初始化一个数据库连接池，也会得到每个进程独立的一组连接。进程数增加时，数据库最大连接数必须重新计算。

---

十一、部署选择：先确认应用协议，再选择服务器

11.1 WSGI 应用

传统 Flask 或 WSGI Django 应用通常使用 WSGI 服务器：

gunicorn myproject.wsgi:application

常见运行结构是：

Gunicorn master
├── worker process 1
├── worker process 2
└── worker process 3

每个 worker 可以采用同步 worker、线程 worker 或其他服务器支持的并发方式。具体参数和行为属于 Gunicorn 的实现，而不是 WSGI 规范保证。

适合 WSGI 的场景包括：

• 主要是普通 HTTP 请求；
• 依赖大量成熟的 WSGI 中间件或扩展；
• 业务代码和依赖库以同步方式为主；
• 没有 WebSocket 或其他事件型长连接需求。

11.2 ASGI 应用

FastAPI、Starlette 等 ASGI 应用可以直接使用 Uvicorn：

uvicorn main:app --host 0.0.0.0 --port 8000

生产环境常见结构是：

客户端
  -> Nginx、云负载均衡或网关
      -> Uvicorn/其他 ASGI server
          -> ASGI 应用

ASGI 服务器负责 HTTP、WebSocket 等协议的解析；FastAPI 负责路由和业务层。ASGI 官方实现列表包含 Uvicorn、Daphne、Hypercorn 等服务器，其中不同服务器支持的 HTTP 版本、WebSocket、HTTP/2 或 HTTP/3 能力并不完全相同，部署时应按实际需求验证，而不能仅凭“支持 ASGI”推断全部协议能力。(asgi.readthedocs.io)

11.3 进程数不是越多越好

worker 数量增加通常可以：

• 利用更多 CPU 核心；
• 隔离单个进程崩溃；
• 提高 CPU 密集型任务的吞吐上限。

但也会增加：

• 内存占用；
• 数据库连接数；
• 缓存副本；
• 启动时间；
• 日志和监控实例数量。

如果每个 worker 建立一个数据库连接池，池大小为 P，worker 数为 W，仅应用进程就可能需要接近：

总连接数 ≈ W × P

还要为管理工具、迁移任务和其他服务保留连接。因此进程数应由 CPU、内存、数据库连接上限和请求类型共同决定。

---

十二、如何判断应该使用 WSGI 还是 ASGI

选择协议时，优先看业务的交互形态，而不是框架流行度。

选择 WSGI 更自然的情况

请求进入
  -> 同步读取数据
  -> 同步执行业务逻辑
  -> 返回完整响应

例如传统后台管理系统、同步 ORM 应用、已有成熟 WSGI 扩展的项目。迁移到 ASGI 并不会自动提升性能；如果所有底层库都是同步阻塞的，ASGI 只是增加了一层线程适配或并发调度。

选择 ASGI 更自然的情况

连接建立
  -> 多次接收事件
  -> 多次发送事件
  -> 长时间保持连接

例如：

• WebSocket；
• Server-Sent Events；
• 长轮询；
• 大量异步 HTTP 请求；
• 异步数据库、异步消息客户端；
• 需要在同一连接上持续收发数据的协议。

ASGI 的优势成立需要满足一个条件：

等待操作必须能够让出事件循环

如果代码中存在大量：

requests.get(...)
time.sleep(...)
同步数据库查询(...)
大规模 CPU 计算(...)

那么仅把函数声明改为 async def 不足以获得 ASGI 的并发收益。

---

十三、常见误解与诊断路径

误解一：async def 就一定更快

错误原因：

async def endpoint():
    result = blocking_library.call()
    return result

诊断方法：

1. 查看事件循环是否出现长时间无响应；
2. 检查同步 HTTP 客户端、数据库驱动和文件操作；
3. 用采样 profiler 或事件循环延迟监控定位阻塞调用；
4. 将真正的异步库与同步库区分开；
5. 必要时用线程隔离同步 I/O。

误解二：ASGI 应用只能部署一个进程

错误原因是把“事件循环”误认为“整个服务只有一个执行单元”。

实际结构可以是：

多个进程 × 每进程一个事件循环 × 每循环多个协程

但每个进程拥有独立内存，不能直接共享普通全局状态。

误解三：WSGI 不能流式响应

WSGI 应用可以返回生成器：

def stream_app(environ, start_response):
    start_response("200 OK", [("Content-Type", "text/plain")])

    def generate():
        yield b"part 1\n"
        yield b"part 2\n"

    return generate()

它可以表达有限的响应流，但 WSGI 缺少 ASGI 那种通用的异步双向事件模型，因此“能流式返回”不等于“能处理 WebSocket”。

误解四：适配器能弥补所有协议差异

WSGI-to-ASGI 适配器可以帮助旧 HTTP 应用运行在 ASGI 服务器中，但它不能把一个不支持 WebSocket 的 WSGI 应用变成 WebSocket 应用。适配器解决的是调用格式和执行环境转换，不是应用能力凭空增加。

误解五：请求断开可以忽略

ASGI HTTP 规范定义了 http.disconnect，并规定连接关闭后调用 send() 可能抛出服务器相关的 OSError；这对长轮询、流式响应和 WebSocket 清理尤其重要。(asgi.readthedocs.io)

长任务或流式响应应考虑：

try:
    while True:
        event = await receive()
        if event["type"] == "http.disconnect":
            break
        # 继续业务处理
except OSError:
    # 客户端已断开，释放资源或取消下游任务
    pass

如果忽略断开事件，应用可能继续查询数据库、读取文件或调用外部服务，最终把结果发送给已经不存在的客户端，造成无效资源消耗。

---

十四、最终的判断框架

可以用下面的顺序做技术判断：

第一步：确认连接模型

只有一次 HTTP 请求和一次响应？
    -> WSGI 或 ASGI 都可以

需要多次收发事件或长连接？
    -> 优先 ASGI

第二步：确认依赖库模型

数据库、HTTP 客户端、消息客户端主要是同步的？
    -> WSGI 通常更直接
    -> ASGI 需要线程隔离同步阻塞操作

底层库原生支持 async/await？
    -> ASGI 能更自然地利用 I/O 并发

第三步：确认迁移成本

已有 Flask/Django + 大量 WSGI 扩展？
    -> 继续使用 WSGI，除非有明确的 ASGI 需求

新建 API + WebSocket/SSE/异步 I/O？
    -> 选择 ASGI

第四步：确认部署约束

必须分别验证：

• 应用服务器是否支持目标协议；
• WebSocket 是否需要额外代理配置；
• 反向代理是否正确转发升级请求；
• worker 数量是否超过数据库连接和内存预算；
• 生命周期初始化是否会在每个进程重复执行；
• 中间件是否与目标协议匹配。

WSGI 的核心是：

同步调用应用，消费返回的响应 iterable

ASGI 的核心是：

异步调用应用，通过 scope、receive 和 send 处理事件

前者把 HTTP 请求—响应周期标准化，后者进一步把连接和协议事件标准化。并发性能、长连接能力和部署方式，最终都可以从这两个调用模型推导出来。

---

系列导航与关联阅读

• 系列入口：Python 完整学习路线：从语言模型、并发到 Web、数据、AI 与生产交付
• 上一篇：Django REST Framework：Serializer、ViewSet、权限、分页和限流
• 下一篇：Python Web API 工程：契约、错误、分页、幂等、限流和版本
• 延伸：FastAPI 完整基础：路由、依赖注入、校验、异步和生命周期
• 延伸：Flask 完整基础：应用工厂、Context、Blueprint、扩展和部署

官方资料

• FastAPI Documentation
• ASGI Specification

本文依据 Python 官方文档、相关 PEP 与生态项目官方文档重新梳理；正文、示例与工程清单由 WR BLOG 编写。