踩坑实录：基于 FastAPI 构建高并发 LLM 流式代理网关（背压控制与连接池调优）

共计 4373 个字符，预计需要花费 11 分钟才能阅读完成。

上个月我们在内网部署基于 vLLM 的本地推理集群，并在前端架设了一套用于 API Key 鉴权、流控计费和日志审计的 Python 反向代理层。本以为只是个简单的 StreamingResponse 透传业务，结果在模拟 300 并发压测时，网关容器的内存一路从 300MB 飙到 8GB 触发 OOM，同时伴随大量 HTTPX 连接池挂死与 504 错误。

本文记录我排查并重构这套大模型流式网关的全过程，核心解决三个生产级硬核问题：客户端中断后的异步任务泄漏 、 慢消费者的背压（Backpressure）失控，以及 HTTPX 连接池复用与 Keep-Alive 调优。

一、痛点拆解：为什么传统的异步流式转发会崩？

很多开发者用 FastAPI 代理 LLM 的 Server-Sent Events (SSE) 流时，写法通常像这样：

# 典型反面教材：极易内存泄漏的流式透传
@app.post("/v1/chat/completions")
async def proxy_chat(request: Request):
    payload = await request.json()
    
    async def event_generator():
        client = httpx.AsyncClient()
        async with client.stream("POST", UPSTREAM_URL, json=payload) as resp:
            async for chunk in resp.aiter_bytes():
                yield chunk

    return StreamingResponse(event_generator(), media_type="text/event-stream")

这段代码在单并发调试时毫无破绽，但放到高并发或弱网移动端场景下，直接暴露出三个致命缺陷：

1. 未感知客户端断开：如果用户在中途点了“停止生成”或关闭了浏览器，下游 TCP 已经断开（RST/FIN），但上面的 generator 会继续向模型推理端拉取完整的几千个 Token，直到模型生成完毕。计算资源与网络带宽被严重浪费。
2. 无内存背压控制（Backpressure）：当推理端以 80 tokens/s 的极高速度输出，而客户端处于 3G 弱网每秒只能接收 10 tokens 时，Python 进程会在内存中无上限缓冲未发送的数据块。并发一高，进程立刻被打爆。
3. 每个请求自建 AsyncClient：频繁建立 / 销毁 SSL 握手与 TCP 连接，瞬时触发 TIME_WAIT 堆积和文件描述符耗尽。

二、架构重构：带背压感知与连接隔离的生产级网关

针对上述问题，我们重构了代理管道。核心逻辑在于：引入全局持久化连接池，并在生成器中注入 request.is_disconnected() 检查与固定大小的异步队列缓冲区。

1. 单例安全客户端与连接池配置

# core/http_client.py
import httpx

class UpstreamClientManager:
    client: httpx.AsyncClient = None

    @classmethod
    def get_client(cls) -> httpx.AsyncClient:
        if cls.client is None or cls.client.is_closed:
            # 严格限制连接上限与空闲回收策略
            limits = httpx.Limits(
                max_connections=500,
                max_keepalive_connections=100,
                keepalive_expiry=30.0
            )
            # 大模型推理 TTFT 往往较高，需细分各类超时时间
            timeout = httpx.Timeout(
                connect=5.0,
                read=60.0,
                write=5.0,
                pool=10.0
            )
            cls.client = httpx.AsyncClient(
                limits=limits,
                timeout=timeout,
                http2=True  # 开启 HTTP/2 多路复用，极大降低长连接握手开销
            )
        return cls.client

    @classmethod
    async def close_client(cls):
        if cls.client and not cls.client.is_closed:
            await cls.client.aclose()

2. 具备主动熔断与背压的 Generator

为了让网关既能抗压，又能在客户端断开时主动向 vLLM 发起中止信号（利用 HTTPX 请求取消通知上游断开生成），我们需要精细化控制数据管道：

# api/routes.py
import asyncio
from fastapi import APIRouter, Request, HTTPException
from fastapi.responses import StreamingResponse
from core.http_client import UpstreamClientManager

router = APIRouter()
UPSTREAM_ENDPOINT = "http://10.0.0.88:8000/v1/chat/completions"

@router.post("/v1/chat/completions")
async def chat_proxy(request: Request):
    payload = await request.json()
    client = UpstreamClientManager.get_client()

    async def stream_wrapper():
        try:
            req = client.build_request("POST", UPSTREAM_ENDPOINT, json=payload)
            resp = await client.send(req, stream=True)
            
            if resp.status_code != 200:
                error_body = await resp.aread()
                yield f"data: {{\"error\": \"Upstream returned {resp.status_code}\"}}\n\n".encode()
                return

            # 主动迭代并监测客户端状态
            async for chunk in resp.aiter_raw():
                # 检查下游是否主动切断连接
                if await request.is_disconnected():
                    # 触发上游连接强行关闭，vLLM 会捕获此中断并停止后续计算
                    await resp.aclose()
                    break
                
                if chunk:
                    yield chunk
                    # 关键让渡：在长连接 IO 密集场景下避免单协程饿死 Event Loop
                    await asyncio.sleep(0)

        except asyncio.CancelledError:
            # 捕获 ASGI 服务器抛出的取消信号
            pass
        finally:
            if 'resp' in locals() and not resp.is_closed:
                await resp.aclose()

    return StreamingResponse(stream_wrapper(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no" # 禁用 Nginx 缓冲区，确保 SSE 流式零延迟输出
        }
    )

三、底层调优：Uvicorn 与 Linux 网络栈魔鬼细节

写完 Python 代码只是完成了一半，在高并发大吞吐的场景下，宿主机系统参数与 ASGI 服务器的启动参数直接决定了网关吞吐量的天花板。

1. 抛弃默认事件循环，强制上 UVLoop

Python 原生自带的 asyncio 事件循环在大量小包频繁吞吐（如 SSE 逐字返回）时 CPU 消耗严重。换用 uvloop 可以带来接近 35%~40% 的吞吐量提升。

pip install uvloop httptools

启动服务时务必明确指定：

uvicorn main:app \
  --host 0.0.0.0 \
  --port 8080 \
  --workers 4 \
  --loop uvloop \
  --http httptools \
  --limit-concurrency 1000 \
  --backlog 2048

2. Linux 内核 TCP 状态调优

压测过程中我们遇到了大量的 Cannot assign requested address 错误，这是端口被未释放的 TIME_WAIT 连接占满导致的。编辑 /etc/sysctl.conf：

# 开启快速回收与复用 TIME-WAIT sockets
net.ipv4.tcp_tw_reuse = 1
# 扩大本地端口范围
net.ipv4.ip_local_port_range = 1024 65535
# 增大 TCP 半连接队列和全连接队列
net.ipv4.tcp_max_syn_backlog = 8192
net.core.somaxconn = 8192
# 降低系统保活探测间隔，更快检测异常断开
net.ipv4.tcp_keepalive_time = 300
net.ipv4.tcp_keepalive_intvl = 15
net.ipv4.tcp_keepalive_probes = 3

执行 sysctl -p 立即生效。

四、压测数据对比：重构前后实测

我们使用 Locust 模拟 300 个并发客户端，每次请求生成约 300 tokens，其中随机注入 20% 的“用户中途断开”操作（模拟跳出率），测试集群配置为 4 核心 / 8GB 内存网关容器 + 2 卡 A100 推理集群。

数据表明，及时的 is_disconnected() 拦截配合流关闭信号，不仅挽救了网关本身的内存，还直接将推理集群从无意义的“鬼影计算”中解脱出来，这在昂贵的 GPU 资源池场景下意义极其重大。

总结与避坑心得

1. 慎用全局中间件缓冲 Body：很多团队喜欢在中间件里做全局的请求日志记录，通过 await request.body() 强行读取流，这会导致 FastAPI 的流式上下文退化成全量缓冲，让 SSE 彻底失效。
2. Nginx 层的暗坑：如果网关前面挂了 Nginx 反向代理，务必显式声明 proxy_buffering off; 和 proxy_read_timeout 600s;，否则不仅流式推送会变成卡顿的“一次性吐出”，连接还会被 Nginx 默认 60s 超时直接掐断。
3. 警惕内存中的字符串拼接：在处理大模型流式响应时，严禁使用 += 拼接大块字符串再处理。坚持使用原始二进制块（aiter_raw() / bytes）直通转发，对需要审计计费的数据，异步投递到后台队列处理，不要阻塞主转发协程。