本文介绍如何使用多线程（`threading`）在 fastapi websocket 应用中长期、独立、可控制地并行执行两个 i/o 密集型阻塞函数（如磁盘读写），避免阻塞事件循环，同时支持客户端指令动态启停。
在构建实时流式服务（如语音转录、日志监控、传感器数据采集）时，常需一对协同工作的阻塞函数：一个持续写入磁盘（foo），另一个轮询读取并转发结果（bar）。由于它们天然依赖文件系统 I/O，强行改造成 async 不仅增加复杂度（需处理异步文件操作或内存共享），还可能引入资源竞争与取消不可靠等问题。此时，多线程是更简洁、可控且符合实际需求的解决方案。
✅ 正确做法：用 threading.Thread + threading.Event 实现可控并发
核心思路是：
- 将 foo 和 bar 分别封装为独立线程；
- 使用 threading.Event 作为线程间通信的“停止信号”；
- 在 WebSocket 连接生命周期内启动/停止线程，并确保资源清理。
以下是一个生产就绪的示例实现（已适配 FastAPI WebSocket 场景）：
import threading
import time
import json
from fastapi import WebSocket, WebSocketDisconnect
from typing import Dict, Optional
# 全局线程管理：按 client_id 存储运行中的线程与控制事件
ACTIVE_THREADS: Dict[str, Dict[str, any]] = {}
def foo(url: str, client_id: str, stop_event: threading.Event):
"""模拟持续写入磁盘的任务（阻塞式）"""
file_path = f"/tmp/{client_id}_output.txt"
counter = 0
while not stop_event.is_set():
try:
# ✅ 模拟耗时 I/O：写入时间戳和计数器
with open(file_path, "a") as f:
f.write(f"[{time.time():.3f}] data_{counter}\n")
counter += 1
time.sleep(2) # 控制写入频率，避免过载
except Exception as e:
print(f"[foo] Error writing for {client_id}: {e}")
break
def bar(client_id: str, stop_event: threading.Event, websocket: WebSocket):
"""模拟持续读取磁盘并推送至 WebSocket 的任务（阻塞式）"""
file_path = f"/tmp/{client_id}_output.txt"
last_size = 0
while not stop_event.is_set():
try:
# ✅ 安全读取：只读新增内容（类似 tail -f）
try:
with open(file_path, "r") as f:
f.seek(0, 2) # 移动到文件末尾
size = f.tell()
if size > last_size:
f.seek(last_size)
new_content = f.read(size - last_size).strip()
last_size = size
if new_content:
# ✅ 同步调用 await —— 必须在主线程中执行！
# 所以 bar 不直接 send，而是通过队列或回调通知主线程
# 这里简化：将数据暂存，由主线程轮询发送（见下文）
pass
except FileNotFoundError:
pass # 文件尚未创建，继续等待
time.sleep(0.5) # 轮询间隔，避免 CPU 空转
except Exception as e:
print(f"[bar] Error reading for {client_id}: {e}")
break
# ⚠️ 关键注意：WebSocket.send_text() 是 async 函数，不能在线程中直接 await
# 因此 bar 线程不应直接调用 websocket.send_text() —— 这会引发 RuntimeError
# 推荐方案：使用 queue.Queue 或 asyncio.Queue 在线程与事件循环间安全传递数据? 常见错误与规避策略
| 问题 | 原因 | 解决方案 |
|---|---|---|
| RuntimeError: await outside async function | 在 bar 线程中直接 await websocket.send_text(...) | ✅ 改用 asyncio.Queue：bar 线程 put_nowait(data)，主线程 await queue.get() 后 send_text |
| 线程无法被及时终止 | stop_event 检查不频繁或被阻塞 I/O 卡住 | ✅ 在 foo/bar 中加入 time.sleep() 或 event.wait(timeout)，确保定期响应中断 |
| 文件读写冲突 | 多线程同时访问同一文件 | ✅ 使用 os.path.getmtime() + seek() 增量读取；或采用临时文件 + 原子重命名（如 write → write.tmp → rename） |
| 线程泄漏 | WebSocket 断开后未显式 join() 或清理 | ✅ 在 finally 块中调用 thread.join(timeout=2) 并删除 ACTIVE_THREADS[client_id] |
✅ 推荐增强结构（含安全通信）
import asyncio
import queue
from asyncio import Queue
# 主线程中初始化（在 websocket_endpoint 内）：
data_queue: Queue[str] = Queue()
# bar 线程中（改为）：
def bar(client_id: str, stop_event: threading.Event, data_queue: Queue):
file_path = f"/tmp/{client_id}_output.txt"
last_size = 0
while not stop_event.is_set():
try:
with open(file_path, "r") as f:
f.seek(0, 2)
size = f.tell()
if size > last_size:
f.seek(last_size)
new_lines = f.read(size - last_size).strip().splitlines()
for line in new_lines:
if line.strip():
asyncio.create_task(data_queue.put(line)) # 安全投递
last_size = size
except Exception as e:
print(f"[bar] Read error: {e}")
time.sleep(0.3)
# 主 WebSocket 循环中（新增消费逻辑）：
async def consume_queue():
while not websocket.closed and not stop_event.is_set():
try:
data = await asyncio.wait_for(data_queue.get(), timeout=1.0)
await websocket.send_text(json.dumps({"type": "data", "payload": data}))
except asyncio.TimeoutError:
continue
except Exception as e:
print(f"[queue consumer] Send error: {e}")
break
# 启动消费者任务（非阻塞）：
consumer_task = asyncio.create_task(consume_queue())
# 停止时：
stop_event.set()
foo_thread.join(timeout=2)
bar_thread.join(timeout=2)
consumer_task.cancel()
await asyncio.gather(consumer_task, return_exceptions=True)✅ 总结
- 不要强行 async 化磁盘 I/O 密集型逻辑：threading 更自然、更稳定、更易调试；
- 必须用 threading.Event 替代全局标志位：确保线程能及时、安全退出；
- 禁止在线程中调用 await：所有异步操作（如 websocket.send_text）必须回到事件循环主线程执行；
- 优先选用 asyncio.Queue 跨线程通信：比 threading.Queue 更契合 async 生态，且无死锁风险；
- 始终 join() 线程并设超时：防止连接异常断开导致后台线程残留。
通过以上设计，你既能保留原有阻塞逻辑的清晰性与可靠性，又能实现毫秒级响应的 WebSocket 实时推送，真正兼顾开发效率与生产健壮性。
