单路由阻塞Google Cloud Run API问题求助

核心原因：异步路由中的同步阻塞代码耗尽线程池资源 #

你遇到的问题本质是FastAPI异步路由内的同步阻塞操作耗尽了实例的线程池，导致所有后续请求无法被处理，最终触发504超时。

• FastAPI的async路由如果执行同步代码（比如你用requests爬取网站、同步调用LLM、spacy的同步处理），会自动将这些同步任务交给后台线程池执行。而Python的线程池默认大小有限（4vCPU实例的默认线程池大小通常在8左右）。
• 你的这个路由单次执行需要1.5分钟，每个请求会占用一个线程池线程直到任务完成。如果多个用户并发调用该路由，很快就会把线程池的所有线程占满。此时，实例无法再处理任何新的请求（不管是该路由还是其他路由），哪怕容器CPU/内存利用率很低——因为线程都被长时间阻塞的任务占用了。
• 当Cloud Run检测到实例无法处理请求时，会自动扩容新实例，但如果该路由的请求持续进来，新扩容的实例也会很快耗尽线程池，最终导致所有实例都无法处理请求，出现全局阻塞。

具体解决方案 #

1. 把同步IO操作改为异步实现 #

针对网络IO密集型任务（爬取网站、调用LLM），改用异步客户端替代同步库：

• 爬取网站：用aiohttp代替requests，避免阻塞线程
• 调用LLM：使用LLM服务商提供的异步SDK（比如OpenAI的openai.AsyncOpenAI），将同步调用改为异步调用

示例（异步爬取网站）：

import aiohttp
from bs4 import BeautifulSoup

async def crawl_website(url: str):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            html = await response.text()
            return BeautifulSoup(html, "html.parser")

2. 用进程池处理CPU密集型任务 #

spacy的文本排序属于CPU密集型操作，受Python GIL限制，用线程无法真正并行。可以用concurrent.futures.ProcessPoolExecutor单独处理，避免阻塞主线程池：

from concurrent.futures import ProcessPoolExecutor
import spacy
import asyncio

nlp = spacy.load("en_core_web_sm")
# 根据CPU核心数设置进程池大小
executor = ProcessPoolExecutor(max_workers=2)

def process_text_sync(text: str):
    doc = nlp(text)
    # 你的文本排序逻辑
    return sorted_results

async def process_text(text: str):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(executor, process_text_sync, text)

3. 调整线程池大小（临时缓解） #

可以通过修改FastAPI的线程池配置，增大线程池容量，但这只是临时缓解，不能从根本解决长时间阻塞的问题：

from fastapi import FastAPI
from starlette.concurrency import run_in_threadpool
import concurrent.futures

# 自定义线程池
custom_executor = concurrent.futures.ThreadPoolExecutor(max_workers=32)

app = FastAPI()

@app.get("/your-slow-route")
async def slow_route():
    # 使用自定义线程池执行同步任务
    result = await run_in_threadpool(custom_executor, your_sync_task)
    return result

4. 异步化整个任务流，用消息队列解耦 #

如果该路由的非实时性允许，建议将整个耗时任务放到Cloud Tasks（GCP的消息队列服务）中处理：

• API收到请求后，立即向Cloud Tasks提交任务，返回一个任务ID给用户
• 后台用单独的服务（或Cloud Run服务）消费任务队列，执行耗时操作
• 用户后续通过任务ID查询处理结果
  这种方式彻底避免了API线程被阻塞，是生产环境处理长耗时任务的最佳实践。

5. 添加限流策略 #

给该耗时路由单独设置限流，比如用FastAPI的slowapi库限制并发请求数，避免瞬间占满资源：

from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded
from fastapi import FastAPI

limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.get("/your-slow-route")
@limiter.limit("5/minute")  # 限制每分钟5个请求
async def slow_route():
    # 你的路由逻辑
    return result

注意事项 #

• 不要在异步路由中长时间运行同步代码，这违背了异步框架的设计初衷
• CPU密集型任务优先用多进程处理，避免GIL的影响
• 生产环境中，长耗时任务尽量用消息队列解耦，保证API的响应性

内容的提问来源于stack exchange，提问作者MercuryProbe