SSE结合SQLAlchemy事件实现实时图表推送的方案合理性咨询

你的方案思路方向是对的（用SSE推送实时数据），但代码里存在几个关键问题，会导致生产环境中出现性能、内存或功能异常：

核心问题点 #

• 重复注册监听器+队列笔误：
  每个客户端连接到/stream时，都会注册一个新的after_insert事件监听器，同时代码里监听器用的是recipe_queue.put()，但定义的队列是data_queue——这会导致两个问题：一是数据库每次插入数据都会触发所有已连接客户端对应的监听器（客户端越多，触发次数越多，直接拖垮数据库操作性能）；二是当前连接的队列根本收不到数据，因为监听器往错的队列里塞数据。

• 监听器未清理：
  客户端断开连接后，你没有移除之前注册的after_insert监听器，这些监听器会一直绑定在Data模型上，随着客户端连接数的增加，内存会持续泄漏，最终导致服务崩溃。

• CPU空转浪费资源：
  while True循环里只是轮询队列是否为空，没有任何等待机制，会让线程一直处于忙碌状态，占用大量CPU资源，尤其是在客户端数量多的时候，服务器负载会急剧升高。

改进后的合理方案 #

针对大量客户端的场景，应该采用全局广播+单监听器的模式，避免重复注册和资源浪费：

优化后的代码示例 #

from fastapi import Request
from fastapi.responses import EventSourceResponse
from sqlalchemy import event
from your_models import Data
import asyncio
from typing import AsyncIterable

# 全局广播队列，所有客户端共享
global_broadcast_queue: asyncio.Queue = asyncio.Queue()

# 只注册一次监听器，程序启动时执行（不要放在请求处理函数里）
def setup_db_listener():
    def after_insert_listener(mapper, connection, target):
        # 把新数据放入全局队列，注意这里用线程安全的方式，因为SQLAlchemy事件可能在非异步线程触发
        asyncio.run_coroutine_threadsafe(global_broadcast_queue.put(target.data), asyncio.get_event_loop())
    
    event.listen(Data, 'after_insert', after_insert_listener)

# 程序启动时初始化监听器
setup_db_listener()

@api_router.get('/stream')
async def message_stream(request: Request) -> AsyncIterable[dict]:
    while True:
        # 客户端断开则退出循环
        if await request.is_disconnected():
            break
        # 等待新数据，无数据时挂起，不会占用CPU
        data = await global_broadcast_queue.get()
        yield {"data": data}

额外注意事项 #

• 连接数限制：SSE是长连接，大量客户端同时连接时，要确保服务器（如Uvicorn）配置了足够的worker进程和连接上限，比如通过--workers 4 --limit-concurrency 1000参数调整。
• 数据去重/过滤：如果不同客户端需要不同的图表数据，不能直接用全局广播，需要增加订阅机制（比如让客户端传递ID，只推送对应的数据）。
• 异常处理：要给await global_broadcast_queue.get()添加超时或异常捕获，避免连接异常时阻塞整个循环。

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