FastAPI集成WebSocket客户端后冻结？如何同时作为WS客户端与服务端运行

• 你当前使用的websocket-client库的WebSocketApp.run_forever()是同步阻塞方法，启动后会持续占用所在线程的全部计算资源
• FastAPI的startup启动事件直接运行在主uvicorn事件循环中，直接调用上述阻塞方法会卡死整个事件循环，导致后续REST请求、FastAPI内置WebSocket服务都无法被调度响应
• 现有代码混用了Python2遗留的线程接口、同步time.sleep逻辑，没有和FastAPI基于asyncio的异步运行模型适配

核心思路：将同步阻塞的WebSocket客户端替换为asyncio原生兼容的异步实现，所有任务和FastAPI的REST接口、WebSocket服务端共用同一个事件循环，不单独占用阻塞线程。

1. 依赖替换 #

废弃同步的websocket-client库，安装和asyncio原生兼容的异步WebSocket实现：
pip install websockets
同步版本的WS客户端为多线程同步场景设计，和asyncio事件循环存在天然的适配冲突，不适合在FastAPI这类异步框架中直接使用

2. 代码改造示例 #

import asyncio
import websockets
from fastapi import FastAPI, WebSocket

app = FastAPI()

# 全局存储外部WS连接实例
external_ws = None
# 存储所有接入本地WS服务端的客户端连接集合
local_ws_clients = set()

# 外部WS客户端长连接协程：负责建连、接收消息、断线重连、逻辑转发
async def external_ws_task():
    global external_ws
    target_uri = "ws://xx.xx.xx.xx:xxxxx/"
    while True:
        try:
            # 建立和外部WS服务的连接
            async with websockets.connect(target_uri) as ws:
                external_ws = ws
                print("已成功接入外部WebSocket服务")
                # 循环接收外部服务推送的消息
                async for raw_msg in ws:
                    # 在此处实现自定义的消息过滤逻辑
                    filtered_msg = raw_msg  # 替换为实际的过滤规则
                    print(f"收到外部WS原始消息，过滤后内容: {filtered_msg}")
                    # 将过滤后的消息批量推送给所有本地接入的WS客户端
                    for client_conn in local_ws_clients:
                        await client_conn.send_text(filtered_msg)
        except Exception as e:
            print(f"外部WS连接中断，5秒后自动重连，错误信息: {e}")
            # 异步等待重连间隔，不阻塞事件循环
            await asyncio.sleep(5)

# FastAPI启动钩子：注册外部WS客户端后台任务
@app.on_event("startup")
async def startup():
    # 以异步后台任务形式启动长连接，不阻塞startup流程
    asyncio.create_task(external_ws_task())

# FastAPI内置的WebSocket服务端接口，供本地客户端接入
@app.websocket("/ws/filter-forward")
async def ws_server_endpoint(websocket: WebSocket):
    await websocket.accept()
    local_ws_clients.add(websocket)
    try:
        # 保持连接存活，直到客户端主动断开
        while True:
            await websocket.receive_text()
    except:
        pass
    finally:
        local_ws_clients.remove(websocket)

3. 关键注意事项 #

• 禁止在事件循环中直接调用阻塞方法：原代码中的同步time.sleep需替换为await asyncio.sleep()，同步阻塞的第三方调用必须做异步适配
• 长运行的后台协程（比如WS长连接）必须通过asyncio.create_task()注册为后台任务，不能直接await阻塞在startup初始化流程中
• 不需要为WS客户端单独开启独立线程、独立事件循环：asyncio单线程调度模型本身可以并发处理REST请求、本地WS客户端连接、外部WS客户端连接，所有协程在同一个循环中调度不会互相阻塞
• 原代码中独立线程实现的心跳逻辑，可直接集成在外部WS协程中：比如新增独立的心跳协程，定期向外部WS服务发送心跳包，无需手动创建线程调度

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