本文详细阐述了如何在fastapi应用中启动并监控外部服务（如java服务）的生命周期。通过结合`asyncio.subprocess_shell`、自定义`asyncio.subprocessprotocol`以及fastapi的`lifespan`事件，我们能够实现对外部服务启动日志的实时监听、状态判断，并利用`asyncio.future`实现异步信号通知，确保fastapi在外部服务完全就绪后才开始处理请求，并在应用关闭时优雅地终止外部服务，同时处理潜在的超时情况。
FastAPI与外部服务集成概述
在构建复杂的微服务或混合技术栈应用时，FastAPI服务可能需要依赖并管理其他外部服务，例如启动一个Java后端服务进行数据处理或业务逻辑。这种场景的核心挑战在于如何异步地启动外部服务，实时监控其启动状态（例如通过日志输出），并在确认服务完全可用后才允许FastAPI应用对外提供服务，同时确保在FastAPI关闭时能够优雅地终止这些外部服务。
最初尝试使用asyncio.subprocess_shell来启动外部进程，并通过一个简单的while循环来检查服务状态，但这会导致FastAPI的事件循环阻塞，使得asyncio.SubprocessProtocol无法处理日志输出，从而造成应用程序冻结。解决此问题的关键在于正确地利用asyncio的异步特性和FastAPI的生命周期管理机制。
解决阻塞问题的初步方案：异步轮询
当使用asyncio.subprocess_shell启动外部进程后，我们需要等待进程输出特定的“启动成功”日志。如果在startup_event中直接使用一个不带await的while循环，如下所示：
# 错误示例：可能导致阻塞# while not protocal.is_startup:# pass
这将导致事件循环无法切换到其他任务，包括处理子进程的输出，从而造成死锁。正确的做法是在循环中引入await asyncio.sleep(0.1)，这会暂停当前协程的执行，并将控制权交还给事件循环，允许它处理其他待定的任务，包括子进程的I/O。
import asyncioimport refrom logging import getLoggerfrom fastapi import FastAPIlogger = getLogger(__name__)app = FastAPI()transport: asyncio.SubprocessTransportprotocol: "MyProtocol" # 提前声明类型，避免循环引用class MyProtocol(asyncio.SubprocessProtocol): startup_str = re.compile(b"Server - Started") # 注意这里正则匹配的是bytes is_startup = False is_exited = False def pipe_data_received(self, fd, data): logger.info(f"Received from fd {fd}: {data.decode().strip()}") # super().pipe_data_received(fd, data) # 通常不需要在子类中调用super() for data processing if not self.is_startup: if re.search(self.startup_str, data): # 直接匹配bytes self.is_startup = True logger.info("External service reported startup success.") def pipe_connection_lost(self, fd, exc): if exc is None: logger.debug(f"Pipe {fd} Closed") else: logger.error(f"Pipe {fd} connection lost: {exc}") super().pipe_connection_lost(fd, exc) def process_exited(self): logger.info("External service process exited.") super().process_exited() self.is_exited = True@app.on_event("startup")async def startup_event(): global transport, protocol loop = asyncio.get_running_loop() # 假设 /start_java_server.sh 是一个可执行脚本，用于启动Java服务 transport, protocol = await loop.subprocess_shell(MyProtocol, "/start_java_server.sh") logger.info("Attempting to start external service...") # 等待外部服务启动完成 while not protocol.is_startup: await asyncio.sleep(0.1) # 允许事件循环处理其他任务 logger.info("External service has successfully started.")@app.on_event("shutdown")async def shutdown_event(): global transport if transport: logger.info("Closing external service transport...") transport.close() # 可以添加一个短暂的等待，确保进程有机会响应关闭信号 # await asyncio.sleep(0.5) logger.info("External service transport closed.")
在这个改进的方案中，await asyncio.sleep(0.1)确保了startup_event协程在等待外部服务启动时不会完全阻塞事件循环，从而允许MyProtocol的pipe_data_received方法被调用并更新is_startup状态。
推荐方案：使用FastAPI Lifespan与asyncio.Future
虽然上述方案解决了阻塞问题，但FastAPI推荐使用lifespan上下文管理器来管理应用程序的生命周期事件，而不是使用已被弃用的@app.on_event装饰器。lifespan提供了更清晰的资源管理和错误处理机制。此外，使用asyncio.Future对象作为异步信号量，可以更优雅地通知服务启动和关闭状态，并能轻松集成超时机制。
1. MyProtocol增强：集成asyncio.Future
我们将修改MyProtocol，使其在初始化时接收started_future和exited_future两个asyncio.Future对象。当检测到特定的启动日志时，设置started_future的结果；当进程退出时，设置exited_future的结果。
import asynciofrom contextlib import asynccontextmanagerimport refrom logging import getLoggerfrom fastapi import FastAPIlogger = getLogger(__name__)# 全局变量用于存储transport和protocol实例transport: asyncio.SubprocessTransport = Noneprotocol: "MyProtocol" = Noneclass MyProtocol(asyncio.SubprocessProtocol): # 构造函数现在接收Future对象 def __init__(self, started_future: asyncio.Future, exited_future: asyncio.Future): self.started_future = started_future self.exited_future = exited_future self.startup_str = re.compile(b"Server - Started") # 匹配bytes def pipe_data_received(self, fd, data): # 记录原始数据，方便调试 logger.info(f"[{fd}] {data.decode(errors='ignore').strip()}") # super().pipe_data_received(fd, data) # 通常不需要在此处调用 # 仅当started_future未完成时才尝试匹配启动字符串 if not self.started_future.done(): if re.search(self.startup_str, data): logger.info("Detected 'Server - Started' in logs.") self.started_future.set_result(True) # 标记启动成功 def pipe_connection_lost(self, fd, exc): if exc is None: logger.debug(f"Pipe {fd} Closed") else: logger.error(f"Pipe {fd} connection lost with exception: {exc}") super().pipe_connection_lost(fd, exc) def process_exited(self): logger.info("External service process exited.") super().process_exited() # 标记进程已退出 if not self.exited_future.done(): self.exited_future.set_result(True)
2. 使用lifespan上下文管理器
lifespan是一个异步上下文管理器，它在FastAPI应用启动前执行yield之前的代码，并在应用关闭后执行yield之后的代码。这非常适合管理外部服务的整个生命周期。
@asynccontextmanagerasync def lifespan(app: FastAPI): global transport, protocol loop = asyncio.get_running_loop() # 创建Future对象用于信号通知 started_future = asyncio.Future(loop=loop) exited_future = asyncio.Future(loop=loop) # 启动外部进程，并将Future对象传递给协议 logger.info("Starting external service via subprocess_shell...") transport, protocol = await loop.subprocess_shell( lambda: MyProtocol(started_future, exited_future), "/start_java_server.sh" # 替换为你的Java启动脚本路径 ) try: # 等待外部服务启动，设置5秒超时 logger.info("Waiting for external service to report startup (timeout: 5s)...") await asyncio.wait_for(started_future, timeout=5.0) logger.info("External service successfully started and ready.") except asyncio.TimeoutError: logger.critical("External service startup timed out!") # 在超时情况下，可以选择关闭传输并退出应用，或进行其他错误处理 if transport: transport.close() raise RuntimeError("External service failed to start within the given timeout.") yield # FastAPI应用开始接收请求 # 应用关闭时执行的代码 logger.info("FastAPI application shutting down. Attempting to gracefully close external service...") if transport: transport.close() # 发送关闭信号给子进程 try: # 等待外部服务进程退出，设置5秒超时 logger.info("Waiting for external service to exit (timeout: 5s)...") await asyncio.wait_for(exited_future, timeout=5.0) logger.info("External service gracefully exited.") except asyncio.TimeoutError: logger.warning("External service did not exit gracefully within the given timeout.") # 如果进程未退出，可以考虑发送更强制的终止信号，或记录警告# 将lifespan上下文管理器注册到FastAPI应用app = FastAPI(lifespan=lifespan)# 示例路由@app.get("/")async def read_root(): return {"message": "Hello from FastAPI, external service is running!"}
关键点解析：
asyncio.Future: 它是asyncio中用于表示未来某个操作结果的对象。通过started_future.set_result(True)，我们可以在MyProtocol中异步地通知lifespan函数，外部服务已经启动。await started_future会阻塞直到set_result被调用。asyncio.wait_for(future, timeout): 这是一个强大的工具，它允许我们等待一个Future对象完成，但会在指定的时间后抛出asyncio.TimeoutError。这对于防止外部服务启动或关闭卡住整个应用至关重要。transport.close(): 这会关闭与子进程的通信管道，并向子进程发送一个终止信号（通常是SIGTERM），允许子进程进行清理并优雅退出。错误处理: 在lifespan中，我们捕获了asyncio.TimeoutError。如果外部服务未能在规定时间内启动，我们可以选择记录错误、关闭传输，甚至终止FastAPI应用启动，以避免服务处于不一致状态。同样，在关闭阶段，我们也可以监控外部服务是否按时退出。日志记录: 使用getLogger(__name__)进行日志记录，有助于跟踪外部服务的状态和调试潜在问题。
总结
通过采用FastAPI的lifespan事件和asyncio.Future的组合，我们构建了一个健壮且异步的机制来管理FastAPI应用中外部服务的生命周期。这种方法不仅解决了初始的阻塞问题，还提供了：
清晰的生命周期管理：利用lifespan统一处理启动前和关闭后的逻辑。非阻塞等待：asyncio.Future和await asyncio.wait_for确保FastAPI应用在等待外部服务时不会阻塞事件循环。状态信号通知：asyncio.Future作为异步信号量，使得外部进程的状态变化能够可靠地通知主应用。超时处理：asyncio.wait_for提供了防止无限等待的机制，增强了应用的健壮性。优雅的关闭：确保在FastAPI应用关闭时，外部服务也能收到终止信号并有机会进行清理。
这种模式是处理FastAPI与其他进程间复杂异步交互的推荐方式，能够有效提升应用的稳定性和可维护性。
以上就是在FastAPI中优雅地管理和监控外部服务的启动与关闭的详细内容，更多请关注创想鸟其它相关文章！
版权声明：本文内容由互联网用户自发贡献，该文观点仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。
如发现本站有涉嫌抄袭侵权/违法违规的内容， 请发送邮件至 chuangxiangniao@163.com 举报，一经查实，本站将立刻删除。
发布者：程序猿，转转请注明出处：https://www.chuangxiangniao.com/p/1380110.html
微信扫一扫 
支付宝扫一扫 
