FastAPI 如何触发 FastAPI/Uvicorn 的优雅停机

在本文中，我们将介绍如何触发 FastAPI/Uvicorn 的优雅停机。当我们需要停止 FastAPI 应用程序时，我们希望尽可能地保证应用程序可以正常关闭，并且处理完当前正在处理的请求。

1. 关闭 FastAPI 和 Uvicorn 服务器

为了实现优雅停机，我们首先需要关闭 FastAPI 和 Uvicorn 服务器。我们可以使用以下方式来完成：

import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/")
def read_root():
    return {"Hello": "World"}

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

当我们使用 ctrl+C 停止运行时，FastAPI 和 Uvicorn 服务器将会立即停止，但这没有给予它们完成当前处理的请求的机会。为了实现优雅停机，我们可以使用 uvicorn.Server 类来创建服务器实例，并使用 Server.install_signal_handlers() 方法来安装信号处理程序。

下面是一个示例：

import signal
import uvicorn
from fastapi import FastAPI

app = FastAPI()
server = uvicorn.Server(uvicorn.Config(app))

@app.on_event("shutdown")
async def shutdown():
    server.should_exit = True

@app.get("/")
def read_root():
    return {"Hello": "World"}

if __name__ == "__main__":
    server.install_signal_handlers()
    uvicorn.run(app, host="0.0.0.0", port=8000)

在上面的示例中，我们使用 @app.on_event("shutdown") 装饰器来定义一个异步函数 shutdown()，并在函数中设置服务器实例的 should_exit 属性为 True。这样，当我们停止运行应用程序时，服务器实例将会收到一个信号来优雅地停止。

2. 处理当前请求

在上述示例中，我们设置了服务器实例的 should_exit 属性为 True，但这并没有立即停止服务器，只是发送了一个信号告知服务器应该停止。为了让服务器能够处理完当前正在处理的请求，我们可以使用协程来延迟停机。

下面是一个示例：

import asyncio
import signal
import uvicorn
from fastapi import FastAPI

app = FastAPI()
server = uvicorn.Server(uvicorn.Config(app))

@app.on_event("shutdown")
async def shutdown():
    server.should_exit = True
    await asyncio.sleep(3)

@app.get("/")
def read_root():
    return {"Hello": "World"}

if __name__ == "__main__":
    server.install_signal_handlers()
    uvicorn.run(app, host="0.0.0.0", port=8000)

在上述示例中，我们在 shutdown() 函数中使用了 asyncio.sleep(3) 来延迟停机3秒钟，以确保当前处理的请求可以完成。你可以根据应用程序的实际情况来调整延迟的时间。

3. 优雅停机的重试策略

在某些情况下，应用程序可能无法在规定的时间内完成当前的处理。为了提高可靠性，我们可以实现一个重试策略，在超过一定时间后强制停止服务器。

下面是一个示例：

import asyncio
import signal
import uvicorn
from fastapi import FastAPI

app = FastAPI()
server = uvicorn.Server(uvicorn.Config(app))

SHUTDOWN_TIMEOUT = 10

@app.on_event("shutdown")
async def shutdown():
    server.should_exit = True
    await asyncio.sleep(3)

async def force_shutdown():
    await asyncio.sleep(SHUTDOWN_TIMEOUT)
    if not server.should_exit:
        server.force_exit = True

@app.get("/")
def read_root():
    return {"Hello": "World"}

if __name__ == "__main__":
    server.install_signal_handlers()
    asyncio.ensure_future(force_shutdown())
    uvicorn.run(app, host="0.0.0.0", port=8000)

在上述示例中，我们定义了一个异步函数 force_shutdown()，该函数在延迟了一定时间后，检查服务器实例的 should_exit 属性，如果服务器还未停止，则将其强制退出。