如何通过FastAPI生命周期事件并发运行FastAPI应用与异步TCP服务器-Python教程-PHP中文网

如何通过FastAPI生命周期事件并发运行FastAPI应用与异步TCP服务器

本教程详细阐述了如何在fastapi应用程序的同一事件循环中，高效地启动并管理自定义异步tcp服务器。核心在于利用fastapi的`lifespan`上下文管理器，确保tcp服务器在应用启动时正确调度为后台任务，并在应用关闭时实现优雅停机，避免了将启动逻辑错误地放置在应用关闭阶段的常见陷阱。

引言

在现代异步应用开发中，我们经常需要将不同的服务组件集成到同一个应用程序中。例如，一个Web API可能需要与一个或多个TCP服务器进行实时数据交互，或者将从TCP连接接收到的数据通过WebSocket广播给前端客户端。FastAPI作为一款高性能的Web框架，天然支持异步操作，为这种集成提供了良好的基础。然而，如何确保自定义的异步任务（如TCP服务器）能够与FastAPI应用一同启动、运行并优雅地关闭，是开发者常遇到的挑战。本教程将深入探讨如何利用FastAPI的生命周期事件（lifespan events）来解决这一问题。

FastAPI生命周期事件与asynccontextmanager

FastAPI提供了一个强大的机制来处理应用程序的启动（startup）和关闭（shutdown）逻辑，即通过lifespan参数与asynccontextmanager装饰器结合使用。asynccontextmanager允许我们定义一个异步的上下文管理器，其核心在于yield关键字。

yield之前的代码： 在asynccontextmanager中，yield关键字之前的代码会在应用程序启动时执行。这是初始化资源、启动后台任务的理想位置。
yield之后的代码： yield关键字之后的代码会在应用程序关闭时执行。这通常用于清理资源、停止后台任务或执行其他关闭逻辑。

理解yield在lifespan中的作用至关重要，它是正确管理后台任务生命周期的关键。

常见问题：后台任务未能启动

许多开发者在尝试将自定义异步任务（如TCP服务器）集成到FastAPI应用时，可能会遇到任务未能启动的问题。这通常是由于对lifespan上下文管理器中yield关键字的误解所导致的。

考虑以下一个尝试启动TCP服务器的错误示例：

# main.py (错误示例)
from fastapi import FastAPI
import asyncio
from contextlib import asynccontextmanager

# 假设 start_tcp_server 是一个异步函数
async def start_tcp_server(port):
    print(f"Attempting to start TCP server on port {port}...")
    # ... TCP server setup logic ...
    await asyncio.sleep(3600) # Simulate a long-running server
    print(f"TCP server on port {port} stopped.")

@asynccontextmanager
async def startup_event(app: FastAPI):
    # 应用程序启动阶段
    print("FastAPI application startup initiated.")
    ports = [8001, 8002, 8003]

    yield # 应用程序现在可以处理请求了

    # 应用程序关闭阶段
    # 错误：将启动TCP服务器的代码放在了这里
    print("FastAPI application shutdown initiated. Starting TCP servers...")
    await asyncio.gather(*(start_tcp_server(port) for port in ports))
    print("All TCP servers started (during shutdown).")

app = FastAPI(lifespan=startup_event)

# ... 其他 FastAPI 路由和 WebSocket 逻辑 ...

登录后复制

在这个示例中，await asyncio.gather(*(start_tcp_server(port) for port in ports)) 这行代码被放置在 yield 之后。这意味着这些TCP服务器的启动逻辑只会在FastAPI应用关闭时才会被触发，而不是在应用启动时。因此，当应用运行时，TCP服务器实际上并未启动，导致外部客户端无法连接。

正确方法：在启动阶段调度后台任务并优雅关闭

要正确地在FastAPI应用启动时运行后台TCP服务器，并在应用关闭时优雅地停止它们，我们需要遵循以下步骤：

在yield之前调度任务： 使用asyncio.create_task()将异步TCP服务器函数包装成任务，并在yield之前启动这些任务。
存储任务引用： 将创建的任务引用存储在一个列表中，以便在关闭时能够管理它们。
在yield之后实现优雅关闭： 在yield之后，即应用程序关闭阶段，向这些后台任务发送停止信号，并等待它们完成清理工作。

为了实现优雅关闭，我们可以引入一个asyncio.Event来作为停止信号。TCP服务器任务会监听这个事件，一旦事件被设置，任务就会开始关闭自身。

1. server.py：TCP服务器逻辑

此文件包含处理单个TCP客户端连接的逻辑，以及启动和管理TCP服务器的函数。我们修改start_tcp_server_task，使其能够响应停止信号。

AppMall应用商店

AI应用商店，提供即时交付、按需付费的人工智能应用服务

查看详情

# server.py
import asyncio
import globals # 假设 globals.py 和 websocket_manager.py 存在

async def handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    """处理单个TCP客户端连接的异步函数。"""
    peername = writer.get_extra_info('peername')
    print(f"TCP client connected: {peername}")
    try:
        while True:
            data = await reader.read(1024)
            if not data:
                break # 客户端断开连接
            decoded_data = data.decode('utf-8')
            print(f"Received from {peername}: {decoded_data}")
            # 将接收到的数据广播到所有连接的WebSocket客户端
            await globals.websocket_manager.broadcast(decoded_data)
    except asyncio.CancelledError:
        print(f"TCP client handler for {peername} was cancelled.")
    except Exception as e:
        print(f"Error handling TCP client {peername}: {e}")
    finally:
        writer.close()
        await writer.wait_closed()
        print(f"TCP client disconnected: {peername}")

async def start_tcp_server_task(port: int, stop_event: asyncio.Event):
    """
    启动一个TCP服务器，并在收到停止信号时优雅关闭。
    :param port: TCP服务器监听的端口。
    :param stop_event: 用于通知服务器停止的 asyncio.Event 对象。
    """
    print(f"Attempting to start TCP server on port {port}...")
    server = await asyncio.start_server(handle_client, '0.0.0.0', port)
    addr = server.sockets[0].getsockname()
    print(f"TCP server listening on {addr}")

    try:
        # 等待停止事件被设置，或者任务被取消
        await stop_event.wait()
    except asyncio.CancelledError:
        print(f"TCP server task for port {port} was cancelled.")
    finally:
        print(f"Closing TCP server on port {port}...")
        server.close() # 关闭服务器
        await server.wait_closed() # 等待服务器完全关闭
        print(f"TCP server on port {port} stopped.")

登录后复制

2. globals.py：全局状态管理

此文件用于存储应用程序的全局状态，例如WebSocket管理器实例。

# globals.py
import threading
from websocket_manager import WebSocketManager # 假设 websocket_manager.py 存在

data_storage = {}
data_lock = threading.Lock()
websocket_manager = WebSocketManager()

登录后复制

3. websocket_manager.py：WebSocket连接管理

此文件包含管理活动WebSocket连接的类。

# websocket_manager.py
from fastapi import WebSocket
from typing import List

class WebSocketManager:
    """管理所有活动WebSocket连接的类。"""
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        """接受新的WebSocket连接并将其添加到活动连接列表。"""
        await websocket.accept()
        self.active_connections.append(websocket)
        print(f"WebSocket connected: {websocket.client}")

    def disconnect(self, websocket: WebSocket):
        """从活动连接列表中移除断开的WebSocket连接。"""
        if websocket in self.active_connections:
            self.active_connections.remove(websocket)
            print(f"WebSocket disconnected: {websocket.client}")

    async def broadcast(self, data: str):
        """向所有活动WebSocket连接广播数据。"""
        disconnected_connections = []
        for connection in self.active_connections:
            try:
                await connection.send_text(data)
            except Exception as e:
                print(f"Error broadcasting to WebSocket {connection.client}: {e}")
                disconnected_connections.append(connection)
        # 移除广播失败的连接
        for connection in disconnected_connections:
            self.disconnect(connection)

登录后复制

4. main.py：FastAPI应用与生命周期管理

此文件包含FastAPI应用定义，以及核心的lifespan上下文管理器，用于正确启动和关闭TCP服务器。

# main.py
import asyncio
from contextlib import asynccontextmanager
from fastapi import FastAPI, WebSocket

import globals
from server import start_tcp_server_task # 导入修改后的TCP服务器启动函数

# 用于存储TCP服务器任务和停止信号的全局变量
tcp_server_tasks = []
tcp_server_stop_event = asyncio.Event()

@asynccontextmanager
async def lifespan_event(app: FastAPI):
    """
    FastAPI应用的生命周期事件管理器。
    在应用启动时启动TCP服务器，在应用关闭时优雅停止它们。
    """
    print("FastAPI application startup initiated.")
    ports = [8001, 8002, 8003]

    # --- 应用启动阶段 (在 yield 之前) ---
    for port in ports:
        # 创建并调度TCP服务器任务
        task = asyncio.create_task(start_tcp_server_task(port, tcp_server_stop_event))
        tcp_server_tasks.append(task)

    print("All TCP server tasks scheduled for startup.")

    yield # 应用程序现在已准备好接收请求

    # --- 应用关闭阶段 (在 yield 之后) ---
    print("FastAPI application shutdown initiated.")

    # 设置停止事件，通知所有TCP服务器任务开始关闭
    tcp_server_stop_event.set()

    # 等待所有TCP服务器任务完成其关闭过程
    # return_exceptions=True 允许 gather 在某个任务失败时继续等待其他任务
    await asyncio.gather(*tcp_server_tasks, return_exceptions=True)

    print("All TCP server tasks gracefully stopped.")
    print("FastAPI application shutdown complete.")

# 使用自定义的 lifespan_event 初始化 FastAPI 应用
app = FastAPI(lifespan=lifespan_event)

@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
    """处理WebSocket连接的端点。"""
    await globals.websocket_manager.connect(websocket)
    try:
        while True:
            # 保持连接活跃，或处理从WebSocket接收到的消息
            # 如果不期望从客户端接收消息，可以只用 await websocket.receive_text()
            # 或 await websocket.receive_bytes() 来保持连接
            message = await websocket.receive_text()
            print(f"Received from WebSocket {websocket.client}: {message}")
            # 示例：将从WebSocket接收到的消息广播回其他WebSocket客户端
            # await globals.websocket_manager.broadcast(f"Echo from WS: {message}")
    except Exception as e:
        print(f"WebSocket Error for {websocket.client}: {e}")
    finally:
        globals.websocket_manager.disconnect(websocket)

登录后复制

运行应用

要运行此FastAPI应用，请使用Uvicorn：

uvicorn main:app --reload

登录后复制

当Uvicorn启动时，你将看到类似以下输出，表明TCP服务器已成功启动：

INFO:     Will watch for changes in these directories: ['/path/to/your/project']
INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [xxxxx] using WatchFiles
FastAPI application startup initiated.
Attempting to start TCP server on port 8001...
TCP server listening on ('0.0.0.0', 8001)
Attempting to start TCP server on port 8002...
TCP server listening on ('0.0.0.0', 8002)
Attempting to start TCP server on port 8003...
TCP server listening on ('0.0.0.0', 8003)
All TCP server tasks scheduled for startup.
INFO:     Started server process [xxxxx]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

登录后复制

现在，你的FastAPI应用（监听8000端口）和三个TCP服务器（监听8001、8002、8003端口）都将并发运行。当TCP客户端连接到8001、8002或8003端口并发送数据时，这些数据将通过WebSocket广播给所有连接到/ws端点的WebSocket客户端。

当你停止Uvicorn进程（例如，按CTRL+C）时，你会看到以下输出，表明TCP服务器正在优雅关闭：

INFO:     Shutting down
FastAPI application shutdown initiated.
Closing TCP server on port 8001...
Closing TCP server on port 8002...
Closing TCP server on port 8003...
TCP server on port 8001 stopped.
TCP server on port 8002 stopped.
TCP server on port 8003 stopped.
All TCP server tasks gracefully stopped.
FastAPI application shutdown complete.
INFO:     Finished server process [xxxxx]

登录后复制

以上就是如何通过FastAPI生命周期事件并发运行FastAPI应用与异步TCP服务器的详细内容，更多请关注php中文网其它相关文章！