异步运行Uvicorn时reload重载选项失效问题求助

Uvicorn的reload功能依赖其CLI启动时的父子进程架构：父进程负责监控文件变化，子进程运行实际的Web服务，当检测到文件修改时父进程会重启子进程。但你通过异步任务直接实例化uvicorn.Server并调用serve()时，会跳过这个父子进程的初始化流程，导致reload=True配置无法生效。

方案一：用多进程分离Uvicorn和Rocketry #

通过多进程分别启动Uvicorn（使用原生reload）和Rocketry，让Uvicorn按照正常的reload流程运行：

import multiprocessing
import uvicorn
from rocketry import Rocketry

# 初始化你的应用
app_rocketry = Rocketry()
app_fastapi = ...  # 替换为你的FastAPI实例

def run_rocketry():
    """启动Rocketry调度器"""
    app_rocketry.run()

if __name__ == "__main__":
    # 启动Rocketry子进程
    rocketry_proc = multiprocessing.Process(target=run_rocketry)
    rocketry_proc.start()
    
    # 用Uvicorn CLI方式启动Web服务，开启reload
    uvicorn.run(
        "app.main:app",  # 替换为你的FastAPI应用路径
        reload=True,
        workers=1,
        loop="asyncio"
    )
    
    # 等待Rocketry进程结束
    rocketry_proc.join()

优点：完全复用Uvicorn原生的reload机制，无需额外编写监控逻辑，实现简单。

方案二：手动实现文件监控与服务重启 #

如果必须用异步任务方式启动，可借助watchdog库手动监控文件变化，触发Uvicorn和Rocketry的重启：

1. 先安装依赖：

pip install watchdog

2. 实现代码：

import asyncio
import uvicorn
from rocketry import Rocketry
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# 初始化应用
app_rocketry = Rocketry()
app_fastapi = ...  # 替换为你的FastAPI实例

# 全局变量保存任务引用
server_task = None
sched_task = None

class FileChangeHandler(FileSystemEventHandler):
    """监控Python文件变化，触发服务重启"""
    def __init__(self, restart_callback):
        self.restart_callback = restart_callback
        self.pending_restart = False

    def on_modified(self, event):
        # 仅监控.py文件，避免频繁触发
        if not event.is_directory and event.src_path.endswith(".py"):
            print(f"检测到文件修改：{event.src_path}，准备重启服务...")
            if not self.pending_restart:
                self.pending_restart = True
                asyncio.create_task(self.restart_callback())

class CustomUvicornServer(uvicorn.Server):
    """自定义Server，确保退出时关闭Rocketry"""
    def handle_exit(self, sig: int, frame) -> None:
        app_rocketry.session.shut_down()
        super().handle_exit(sig, frame)

async def start_services():
    """启动Uvicorn和Rocketry服务"""
    global server_task, sched_task
    # 启动Uvicorn
    server = CustomUvicornServer(
        config=uvicorn.Config(
            app_fastapi,
            workers=1,
            loop="asyncio"
        )
    )
    server_task = asyncio.create_task(server.serve())
    # 启动Rocketry
    sched_task = asyncio.create_task(app_rocketry.serve())
    # 等待任一服务结束
    await asyncio.wait([server_task, sched_task], return_when=asyncio.FIRST_COMPLETED)

async def restart_services():
    """重启Uvicorn和Rocketry服务"""
    global server_task, sched_task, handler
    # 标记允许下次重启
    handler.pending_restart = False
    
    # 关闭现有服务
    app_rocketry.session.shut_down()
    if not server_task.done():
        server_task.cancel()
    if not sched_task.done():
        sched_task.cancel()
    
    # 处理任务取消异常
    try:
        await server_task
        await sched_task
    except asyncio.CancelledError:
        pass
    
    # 重启服务
    await start_services()

async def main():
    global handler
    # 启动初始服务
    await start_services()
    
    # 初始化文件监控
    handler = FileChangeHandler(restart_services)
    observer = Observer()
    # 监控当前目录下所有Python文件（递归）
    observer.schedule(handler, path=".", recursive=True)
    observer.start()
    
    # 保持主进程运行
    while True:
        await asyncio.sleep(1)

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        # 手动退出时关闭Rocketry
        app_rocketry.session.shut_down()

优点：保持异步启动的架构，适合需要更精细控制服务生命周期的场景；缺点：需要自己处理重启逻辑，复杂度较高。

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