Uvicorn 优雅停机与工作进程

示例说明

滚动发布时容器收到 SIGTERM，粗暴退出会让在途请求半途而废。timeout_graceful_shutdown 定义宽限窗口：uvicorn 先停止接受新连接，等待存量请求完成，超时才强制断开。应用侧要配套两件事——健康端点切换到排空状态让负载均衡摘流量，后台循环监听关闭事件有序退出而非被硬杀。

技术要点

• 宽限窗口覆盖排空时长上限
• 健康检查反映排空状态引导摘流
• 关闭事件协调后台循环退出

代码解读

health 端点按关闭事件返回不同状态；shutdown 钩子模拟资源释放耗时。background_sync 循环每五秒打点，事件置位后自然退出；runner 用上下文管理器演示取消与等待的规范组合——cancel 后 suppress 取消异常再 await 回收任务。uvicorn.run 传入关键参数。

使用建议

宽限期要小于编排器的终止宽限，给强制退出留余量。排空期间可拒绝新任务提交。多 worker 模式下每个进程独立走这套流程。

代码示例

import asyncio
import contextlib

from fastapi import FastAPI

app = FastAPI()
shutdown_event = asyncio.Event()


@app.get('/health')
async def health():
    if shutdown_event.is_set():
        return {'status': 'draining'}
    return {'status': 'ready'}


@app.on_event('shutdown')
async def release_resources():
    print('draining in-flight requests')
    await asyncio.sleep(0.2)
    print('background loops cancelled')


async def background_sync():
    while not shutdown_event.is_set():
        await asyncio.sleep(5)
        print('sync tick')
    print('background loop exited cleanly')


@contextlib.asynccontextmanager
async def runner():
    task = asyncio.create_task(background_sync())
    try:
        yield
    finally:
        shutdown_event.set()
        task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await task


if __name__ == '__main__':
    # 关键在于命令行参数：
    # uvicorn app:app --timeout-graceful-shutdown 20
    # 收到 SIGTERM 后：停止接新请求 → 排空在途请求 → 最多等 20 秒 → 强制退出
    import uvicorn

    uvicorn.run(
        'app:app',
        host='0.0.0.0',
        port=8000,
        timeout_graceful_shutdown=20,
        timeout_keep_alive=5,
    )