本文介绍如何利用 asyncio 同步原语（如 Event）替代轮询机制，在 WebSocket 收到停止请求时真正实现异步任务的即时取消，避免 generate_image 等长耗时调用被动等待完成。
本文介绍如何利用 asyncio 同步原语（如 event）替代轮询机制，在 websocket 收到停止请求时真正实现异步任务的即时取消，避免 `generate_image` 等长耗时调用被动等待完成。
在异步 Web 服务（如基于 FastAPI + WebSockets 的图像生成接口）中，一个常见痛点是：当客户端主动断开或发送停止信号（如 websocket.stop_requested = True）后，后端仍在执行耗时的异步操作（例如调用 OpenAI DALL·E API），导致资源浪费与响应延迟。你当前的代码通过多次手动检查 stop_requested 标志来“事后”退出，但无法中断正在执行的 await client.images.generate(...) —— 因为该调用本身不响应外部取消信号。
根本解法不是轮询，而是让异步任务能被外部主动取消。Python asyncio 提供了优雅的同步原语，其中 asyncio.Event 是最推荐的方式：
✅ 推荐方案：使用 asyncio.Event 实现可中断等待
首先，在 WebSocket 连接生命周期内初始化一个共享的 Event 对象（建议绑定到 websocket 实例或上下文管理器中）：
WebSocket 8.18.2 是该协议规范的一个重要迭代版本，主要优化了连接稳定性与数据传输效率。它通过全双工通信机制，允许客户端与服务器在单一长连接上实时交换数据，大幅降低传统 HTTP 轮询的开销。该版本增强了心跳保活、自动重连及二进制帧传输能力，适用于即时通讯、在线游戏及金融行情推送等低延迟场景，为开发者提供更可靠的实时网络交互基础。
# 初始化（例如在 WebSocket 连接建立时） websocket.stop_event = asyncio.Event()
然后重构你的 process 方法，将 generate_image 封装为独立任务，并使用 asyncio.wait() 或 asyncio.create_task() 配合 wait_for + timeout=None 实现可取消等待：
class ImageGeneration(BaseAIGeneration):
async def process(self, websocket, prompt):
# 1. 立即响应已停止状态
if websocket.stop_requested:
return None
await super().process(websocket, prompt)
# 2. 创建生成任务
gen_task = asyncio.create_task(
self.generate_image(prompt=prompt, model=self.model, size=size)
)
# 3. 等待任务完成 或 stop_event 被触发（任一先发生）
done, pending = await asyncio.wait(
[gen_task, websocket.stop_event.wait()],
return_when=asyncio.FIRST_COMPLETED
)
# 4. 清理：若 stop_event 先触发，则取消生成任务
if websocket.stop_event.is_set():
gen_task.cancel()
try:
await gen_task # 消费 CancelledError，确保清理完成
except asyncio.CancelledError:
pass
return None
# 5. 否则获取结果
try:
response = gen_task.result()
if not response or not isinstance(response, list):
return None
image_url = response[0].url
return image_url
except Exception as e:
print(f"Image generation failed: {e}")
return None同时，确保在 WebSocket 断开或收到停止指令时正确触发事件：
# 在处理 WebSocket 消息的主循环中（例如 receive loop）
if data.get("action") == "stop":
websocket.stop_requested = True
websocket.stop_event.set() # ⚡ 关键：立即唤醒所有等待者⚠️ 注意事项与最佳实践
-
不要手动轮询
stop_requested：你原设想的while not task.done(): sleep(0.1)不仅低效、增加延迟，还违背 asyncio “等待驱动”设计哲学，且易引发竞态。 -
EventvsFuture：asyncio.Event语义更清晰（表示“某事已发生”），比裸Future更易维护；除非有特殊调度需求，否则优先选Event。 -
API 层需支持取消：上述方案能取消等待，但最终是否能中止 HTTP 请求取决于底层客户端（如
httpx.AsyncClient默认支持 cancellation，而某些 SDK 可能不透传）。建议使用支持 cancellation 的 HTTP 客户端，并确认client.images.generate内部会响应asyncio.CancelledError。 -
异常处理必做：
task.cancel()会向协程抛出CancelledError，务必用try/except捕获并完成资源清理（如关闭流、释放锁）。
✅ 总结
真正的“立即停止”不依赖于高频轮询，而在于构建可协作取消（cooperative cancellation）的异步流程：用 asyncio.Event 作为外部控制信号，配合 asyncio.wait 实现多路等待，再通过 task.cancel() 主动中断挂起的协程。这不仅提升响应性，也使代码更健壮、可测试、符合 asyncio 最佳实践。
