Django中如何仅在POST请求触发时广播Server-Sent Events

一、正确实现思路 #

你的核心需求是用户发起POST请求时，向所有连接的SSE客户端广播事件，最合理的方式是采用订阅-发布模式，维护活跃客户端的连接队列，POST请求触发时直接向所有队列推送消息，避免无效轮询。以下是基于Django的实现示例：

import asyncio
from django.http import StreamingHttpResponse, JsonResponse
import random

# 存储所有活跃SSE客户端的消息队列
active_connections = set()
# 线程锁，保证多协程/线程下操作集合的安全性
connection_lock = asyncio.Lock()

async def client_event_stream():
    """单个客户端的SSE事件生成器"""
    msg_queue = asyncio.Queue()
    
    # 将当前客户端的队列加入活跃连接集合
    async with connection_lock:
        active_connections.add(msg_queue)
    
    try:
        while True:
            # 等待队列中的广播消息
            data = await msg_queue.get()
            yield f'data: {data},SUCCESS\n\n'
            msg_queue.task_done()
    finally:
        # 客户端断开连接时，从集合中移除队列
        async with connection_lock:
            active_connections.remove(msg_queue)

def sse_connect(request):
    """SSE客户端连接入口"""
    return StreamingHttpResponse(client_event_stream(), content_type='text/event-stream')

def trigger_broadcast(request):
    """处理用户POST请求，触发全局广播"""
    success = random.choice([True, False])
    broadcast_content = f"{random.choice(range(1,40))},{success}"
    
    # 异步执行广播任务，不阻塞POST请求响应
    asyncio.create_task(broadcast_to_all_clients(broadcast_content))
    
    return JsonResponse({'success': success})

async def broadcast_to_all_clients(content):
    """向所有活跃SSE客户端推送消息"""
    async with connection_lock:
        # 复制队列集合，避免遍历过程中集合结构变化
        client_queues = list(active_connections)
    
    # 向每个客户端的队列推送消息
    for queue in client_queues:
        await queue.put(content)

这个方案的核心优势：

• 无轮询开销：客户端只在有消息时才推送，避免空转浪费CPU资源
• 线程安全：用异步锁保证集合操作的原子性，不会出现数据竞争问题
• 实时性高：POST请求触发后立即推送消息，无延迟
• 自动清理：客户端断开连接时自动移除队列，避免无效资源占用

二、你提出的全局集合轮询方案的性能问题 #

你给出的方案通过全局set存储数据，SSE客户端每隔0.1秒轮询集合，存在以下明显问题：

1. CPU资源浪费：每个客户端都会周期性轮询集合，当客户端数量较多时，会产生大量无意义的检查操作，占用大量CPU资源。
2. 线程安全隐患：全局set未加锁，多线程/协程同时执行add和pop操作时，可能出现数据丢失、KeyError异常。
3. 消息丢失风险：多个客户端同时轮询时，可能争抢同一条数据，导致部分客户端收不到广播事件。
4. 实时性差：轮询间隔0.1秒意味着消息最多延迟0.1秒才能到达客户端，无法满足低延迟需求。

三、生产环境注意事项 #

• 心跳包：定期向客户端发送:\n\n格式的心跳包，避免网关或浏览器主动断开长连接
• 多进程部署：如果用多进程部署Django，全局集合无法跨进程共享，需要用Redis Pub/Sub等分布式消息队列实现跨进程广播

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