在 Flask 同步路由中调用 WebSocket 发送请求后，需阻塞等待异步 on_message 的响应结果；本文详解两种线程安全方案：asyncio.run_coroutine_threadsafe 直接获取返回值，或配合 queue.Queue 实现跨事件循环通信。
在 flask 同步路由中调用 websocket 发送请求后，需阻塞等待异步 `on_message` 的响应结果；本文详解两种线程安全方案：`asyncio.run_coroutine_threadsafe` 直接获取返回值，或配合 `queue.queue` 实现跨事件循环通信。
在混合同步（Flask）与异步（WebSocket + asyncio）架构中，常见痛点是：Flask 路由必须返回 HTTP 响应，但业务逻辑依赖 WebSocket 的双向通信——即发送请求后需等待服务端通过 on_message 回传结果。由于 Python 的 await 只能在协程中使用，无法在普通同步函数中直接 await，强行调用 asyncio.run() 会启动新事件循环、阻塞主线程且开销大，不适用于高并发 Web 服务。
✅ 推荐方案一：asyncio.run_coroutine_threadsafe（推荐用于可修改 WebSocket 发送逻辑的场景）
该方法将协程提交至已运行的事件循环（如 WebSocket 服务所用的 loop），并返回 concurrent.futures.Future，主线程可安全 .result() 阻塞等待（支持超时）。关键前提是：你持有该事件循环的引用（通常在 WebSocket 初始化时保存）。
import asyncio
from flask import Flask, request, jsonify
app = Flask(__name__)
# 假设这是你的 WebSocket 客户端实例（如 websockets.connect 返回的对象）
ws_client = None
ws_loop = None # 必须在 WebSocket 连接建立时保存 event loop 引用
# 示例：WebSocket 发送并等待响应的协程（需自行实现响应匹配逻辑）
async def send_and_wait(ws, data, timeout=10):
# 1. 发送消息（假设 ws.send 是协程）
await ws.send(json.dumps(data))
# 2. 使用 asyncio.wait_for 等待 on_message 中存入的响应
# （实际需配合唯一 request_id + 字典缓存或 asyncio.Queue 实现）
try:
# 此处简化：假设有一个全局 asyncio.Queue 存储响应
response = await asyncio.wait_for(response_queue.get(), timeout=timeout)
return response
except asyncio.TimeoutError:
raise TimeoutError("WebSocket response timeout")
@app.route("/request", methods=["POST"])
def manageRequest():
data = request.get_json()
# ✅ 提交协程到已运行的事件循环（非新建！）
future = asyncio.run_coroutine_threadsafe(
send_and_wait(ws_client, data),
ws_loop # ← 必须是正在运行的 loop 实例
)
try:
# 阻塞等待结果（可加 timeout 避免永久挂起）
response = future.result(timeout=15) # 单位：秒
result = makeSomething(response)
return jsonify({"success": True, "data": result})
except TimeoutError:
return jsonify({"error": "WebSocket timeout"}), 504
except Exception as e:
return jsonify({"error": str(e)}), 500
⚠️ 注意事项：
WebSocket 8.18.2 是该协议规范的一个重要迭代版本，主要优化了连接稳定性与数据传输效率。它通过全双工通信机制，允许客户端与服务器在单一长连接上实时交换数据，大幅降低传统 HTTP 轮询的开销。该版本增强了心跳保活、自动重连及二进制帧传输能力，适用于即时通讯、在线游戏及金融行情推送等低延迟场景，为开发者提供更可靠的实时网络交互基础。
- ws_loop 必须是 WebSocket 所在线程中已启动且持续运行的 asyncio.AbstractEventLoop（可通过 asyncio.get_running_loop() 在协程内获取并导出）。
- send_and_wait 协程内部需妥善处理请求-响应匹配（例如为每次请求生成唯一 request_id，on_message 解析后存入 asyncio.Queue 或字典缓存）。
- 切勿在 Flask 视图中调用 asyncio.run() —— 它会创建/关闭全新事件循环，严重降低性能且无法复用 WebSocket 连接。
✅ 替代方案二：queue.Queue（适用于无法修改 WebSocket 协程返回逻辑的场景）
当 on_message 已固定为回调式处理，且不便改造成返回值协程时，可用线程安全队列中转响应：
import queue
import threading
# 全局响应队列（线程安全）
response_queue = queue.Queue(maxsize=1)
# 在 on_message 中（运行于 asyncio loop 线程）：
async def on_message(message):
data = json.loads(message)
# ... 处理逻辑
result = await doStuff(data)
# ✅ 从 asyncio 线程向主线程传递数据
try:
response_queue.put_nowait(result) # 非阻塞写入
except queue.Full:
pass # 或记录告警
# Flask 路由中：
@app.route("/request", methods=["POST"])
def manageRequest():
data = request.get_json()
# 1. 发送 WebSocket 请求（同步调用，假设 ws.send 是同步方法）
ws.send(json.dumps(data))
# 2. 阻塞等待响应（可设 timeout）
try:
response = response_queue.get(timeout=10) # 单位：秒
result = makeSomething(response)
return jsonify({"success": True, "data": result})
except queue.Empty:
return jsonify({"error": "No response from WebSocket"}), 504
? 总结建议：
- 优先选择方案一：它更符合 asyncio 设计哲学，响应流清晰可控，且避免全局状态。
- 务必管理好事件循环生命周期：确保 ws_loop 是长期运行的（如通过 asyncio.create_task() 启动后台任务维持）。
- 始终设置超时：防止因网络异常或服务端无响应导致 Flask 请求永久挂起。
- 生产环境注意线程安全：queue.Queue 天然线程安全；若用 asyncio.Queue，则必须在同一线程（即事件循环线程）中操作，不可跨线程 .get()。
通过上述任一方式，即可在保持 Flask 同步接口的同时，安全、高效地桥接 WebSocket 异步通信，实现真正的“请求-响应”闭环。
