如何在指定时间后终止FastAPI端点执行以降低CPU资源开销

使用场景 #

调用/do_something的客户端微服务在request/post()调用中设置了60秒固定超时，且无法修改。如果/do_something接口执行耗时长达10分钟，客户端在60秒后就会停止等待响应，但该接口仍会浪费10分钟的CPU资源，增加运维成本，而我们的预算十分有限。

当前代码 #

import time
from uvicorn import Server, Config
from random import randrange
from fastapi import FastAPI

app = FastAPI()

def some_func(text):
    """
    计算密集型函数，执行时间取决于输入文本大小
    """
    randinteger = randrange(1,120)
    time.sleep(randinteger)# 模拟文本处理耗时
    return text


@app.get("/do_something")
async def do_something():
    response = some_func(text="hello world")
    return {"response": response}

# 启动服务
if __name__ == '__main__':
    server = Server(Config(app=app, host='0.0.0.0', port=3001))
    server.run()

期望解决方案 #

• /do_something接口需在60秒后终止当前请求处理，释放资源以处理下一个请求
• 接口因超时时需记录自定义日志信息
• 操作不能杀死服务进程，且需支持多线程/多进程模式

尝试的代码及问题 #

我尝试了以下代码，但超时发生时服务器会被直接杀死，求解决办法：

import logging
import time
import timeout_decorator
from uvicorn import Server, Config
from random import randrange
from fastapi import FastAPI

app = FastAPI()


@timeout_decorator.timeout(seconds=2, timeout_exception=StopIteration, use_signals=False)
def some_func(text):
    """
    计算密集型函数，执行时间取决于输入文本大小
    """
    randinteger = randrange(1,30)
    time.sleep(randinteger)# 模拟文本处理耗时
    return text


@app.get("/do_something")
async def do_something():
    try:
        response = some_func(text="hello world")
    except StopIteration:
        logging.warning(f'Stopped /do_something > endpoint due to timeout!')
    else:
        logging.info(f'(  Completed < /do_something > endpoint')

    return {"response": response}


# 启动服务
if __name__ == '__main__':
    server = Server(Config(app=app, host='0.0.0.0', port=3001))
    server.run()

问题根源在于直接在异步接口中调用同步阻塞函数，且使用的超时装饰器机制可能影响服务主进程。正确的做法是将耗时同步任务放到独立线程/进程池执行，通过安全的超时控制终止任务，同时不影响服务本身。

以下是修正后的代码：

import logging
import time
from concurrent.futures import ThreadPoolExecutor, TimeoutError
from uvicorn import Server, Config
from random import randrange
from fastapi import FastAPI, HTTPException

app = FastAPI()
# 创建线程池，可根据服务负载调整线程数量
executor = ThreadPoolExecutor(max_workers=4)

# 配置日志输出
logging.basicConfig(level=logging.INFO)

def some_func(text):
    """
    计算密集型函数，执行时间取决于输入文本大小
    """
    randinteger = randrange(1, 120)
    time.sleep(randinteger)  # 模拟文本处理耗时
    return text


@app.get("/do_something")
async def do_something():
    try:
        # 在线程池中执行同步函数，设置60秒超时
        response = await app.state.executor.submit(some_func, "hello world").result(timeout=60)
        logging.info(f"完成 /do_something 接口请求")
        return {"response": response}
    except TimeoutError:
        logging.warning(f"因超时终止 /do_something 接口请求")
        raise HTTPException(status_code=408, detail="请求超时（60秒）")
    except Exception as e:
        logging.error(f"/do_something 接口执行出错: {str(e)}")
        raise HTTPException(status_code=500, detail="服务器内部错误")


# 启动服务
if __name__ == '__main__':
    # 将线程池挂载到应用状态，方便接口调用
    app.state.executor = ThreadPoolExecutor(max_workers=4)
    server = Server(Config(app=app, host='0.0.0.0', port=3001))
    try:
        server.run()
    finally:
        # 服务停止时关闭线程池，避免资源泄漏
        app.state.executor.shutdown()

方案说明 #

1. 线程池隔离任务：用ThreadPoolExecutor将同步耗时任务放到独立线程执行，避免阻塞FastAPI的异步事件循环，同时支持多线程并发处理请求。
2. 安全超时控制：通过result(timeout=60)设置超时，超时后抛出TimeoutError，仅终止当前任务，不会影响服务主进程。
3. 完善日志与响应：针对成功、超时、异常场景分别记录日志，并返回标准HTTP状态码，符合REST接口规范。
4. 资源清理：服务停止时关闭线程池，避免资源泄漏。

如果some_func是CPU密集型任务，推荐替换为ProcessPoolExecutor（进程池），规避Python GIL限制，代码逻辑无需大幅修改。

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