# dj_sse

**Repository Path**: duans/dj_sse

## Basic Information

- **Project Name**: dj_sse
- **Description**: No description available
- **Primary Language**: Unknown
- **License**: Not specified
- **Default Branch**: master
- **Homepage**: None
- **GVP Project**: No

## Statistics

- **Stars**: 0
- **Forks**: 0
- **Created**: 2026-03-16
- **Last Updated**: 2026-03-19

## Categories & Tags

**Categories**: Uncategorized

**Tags**: None

## README

# SSE(server side event)服务端事件

## websocket和sse的区别

![see和websocket](media/see和websocket.jpeg)

## SSE实现服务端消息推送

1. **单向性匹配**：你的需求是“服务端推给前端”，不需要前端频繁发数据给服务端（除了心跳或重连），这完美契合 SSE 的单向特性。
2. **自动重连**：网络波动导致连接断开时，浏览器会自动尝试重连（这是 `EventSource` 的内置功能），无需你写复杂的断线重连逻辑。
3. **开发简单**：前端只需几行 JS，后端只需返回一个流式响应，不需要像 WebSocket 那样处理握手、帧解析等复杂协议。
4. **资源消耗低**：相比 WebSocket，SSE 基于 HTTP，对服务器并发连接的开销略小（取决于具体实现），且更容易穿过防火墙和代理。

### 流式数据接口

服务端

```python
# views.py
from django.http import StreamingHttpResponse
# 定义流式接口
def stream(request):
    import time
    # 定义生成器函数, 不断生成数据
    def generated_result():
        # 实际场景: 查询数据库中的最新数据
        time.sleep(0.5) 
        # SSE要求的数据格式 data: 响应数据
        yield f'data: {json.dumps({"message": "有新订单了-"+str(randint(1000,9999))})}\n\n'

    # 创建流式响应对象: 需要传入生成器函数, 设置content_type响应头为text/event-stream
    response=StreamingHttpResponse(generated_result(),content_type='text/event-stream')
    # 设置响应数据格式: sse需要的数据格式
    # resposne['content_type']="text/event-stream"
    # 禁止浏览器缓存
    response['Cache-Control'] = 'no-cache'
    # 禁止Nginx缓存
    response['X-Accel-Buffering'] = 'no' 
    # 返回响应对象
    return response
```

客户端

```js
// 创建eventSource对象, 和服务端建立链接, 默认get请求方式
event=new EventSource('http://localhost:8000/stream/')
// 监听服务端的消息
event.onmessage = function(event) {
    // 接受服务端响应的流式数据, 并将其转换为js对象
    const data = JSON.parse(event.data);
};
// 监听链接中断事件
event.onerror = function(err) {
    console.log('err')
};
```

## 完整实现方案 (Django + 前端)

我们将实现一个功能：当数据库产生新订单时，后台管理页面立即弹出通知。

### 1. 核心架构设计

- **推送通道**：前端通过 `EventSource` 连接 Django 的 SSE 视图。
- 触发机制
  - **方案 A (推荐 - 生产级)**：订单创建后，发送消息到 **Redis Pub/Sub** 或 **Celery Signal**。SSE 视图订阅 Redis 频道，一有消息立刻推送。
  - **方案 B (简单 - 演示级)**：SSE 视图轮询数据库（不推荐，有延迟且浪费资源）。
  - **方案 C (Django Channels)**：使用 Django Channels 的 Group 功能（功能最强，但部署复杂）。

**下面采用最通用且高效的“方案 A：Redis Pub/Sub”模式进行演示。**

### 2. 后端实现 (Django + Redis)

首先安装依赖：

```
pip install redis
```

**views.py**:

```python
from django.shortcuts import render,redirect
import json
import redis
from django.http import StreamingHttpResponse
from django.conf import settings
from datetime import datetime
from random import randint
from django.http import HttpResponse,HttpResponseRedirect
from django.views import View
import uuid

# 初始化 Redis 连接
r = redis.Redis(host='localhost', port=6379, db=0)

def index(request):    
    return redirect('order_list')

def order_list(request):
    return render(request, 'order_list.html')



class Publish(View):
    def get(self,request):
        return render(request,'publish.html')
    
    def post(self,request):
        # body=json.loads(request.body.decode())
        body=request.POST
        payload = {
            "type": "new_order",
            "order_id": uuid.uuid4().hex,
            "customer": body.get('customer','匿名'),
            "amount": body.get('amount',0),
            "time": datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        }
        # 发布订单消息到redis的new_orders频道
        r.publish('new_orders', json.dumps(payload))
        # 可以传递路径地址, 也可以传递路由name
        # return redirect('publish')
        # 必须传路径地址
        return HttpResponseRedirect('/publish/')

# 监听redis channel消息
def subscribe(request):
    """
    SSE 视图：监听 Redis 频道，有新订单时推送
    """
    user_id = request.user.id
    # 每个用户或管理员组可以订阅不同的频道，这里假设所有管理员订阅 'new_orders'
    channel = "new_orders"
    
    def event_generator():
        # 创建对象
        pubsub = r.pubsub()
        # 监听频道
        pubsub.subscribe(channel)
        
        try:
            # 监听 Redis 消息
            for message in pubsub.listen():
                if message['type'] == 'message':
                    # 将字节串转换为字符串
                    data = message['data'].decode('utf-8')
                    # 格式化为 SSE 标准格式
                    yield f"data: {data}\n\n"
                elif message['type'] == 'subscribe':
                    # 订阅成功，可以发个欢迎消息
                    yield f'data: {{"message": "连接成功，正在监听新订单..."}}\n\n'
                    
        except (BrokenPipeError, ConnectionAbortedError, KeyboardInterrupt):
            print(f"用户 {user_id} 断开连接")
        finally:
            pubsub.unsubscribe(channel)
            pubsub.close()

    response = StreamingHttpResponse(
        event_generator(),
        content_type='text/event-stream'
    )
    # 关键配置：禁止缓存，防止 Nginx/浏览器缓存导致收不到实时消息
    response['Cache-Control'] = 'no-cache'
    response['X-Accel-Buffering'] = 'no' 
    # response['Connection'] = 'keep-alive'
    
    return response
```

### 3. 前端实现 (HTML + JavaScript)

在后台管理系统的模板中加入以下代码：

```html
<!doctype html>
<html>
  <head>
    <title>订单管理后台</title>
    <style>
      #notification-area {
        position: fixed;
        top: 20px;
        right: 20px;
        width: 300px;
        z-index: 1000;
      }
      .toast {
        background: #fff;
        border-left: 5px solid #28a745;
        box-shadow: 0 2px 10px rgba(0, 0, 0, 0.2);
        padding: 15px;
        margin-bottom: 10px;
        border-radius: 4px;
        animation: slideIn 0.5s ease-out;
      }
      @keyframes slideIn {
        from {
          transform: translateX(100%);
          opacity: 0;
        }
        to {
          transform: translateX(0);
          opacity: 1;
        }
      }
      .tb-order {
        border: 1px solid #ddd;
        width: 100%;
        border-collapse: collapse;
      }
      .tb-order td,
      .tb-order th {
        border: 1px solid #ddd;
        text-align: center;
      }
      .tb-order .thead {
        background-color: #eee;
        height: 36px;
      }
    </style>
  </head>
  <body>
    <h1>订单管理列表</h1>
    <a href="/publish/" target="_blank">用户提交订单</a>
    <hr />
    <!-- 通知容器 -->
    <div id="notification-area"></div>

    <table class="tb-order">
      <thead>
        <tr class="thead">
          <th>订单编号</th>
          <th>下单用户</th>
          <th>订单金额</th>
          <th>下单时间</th>
        </tr>
      </thead>
      <tbody id="tbody">
        <!-- <tr>
          <td>订单编号</td>
          <td>下单用户</td>
          <td>订单金额</td>
          <td>下单时间</td>
          <td>操作</td>
        </tr> -->
      </tbody>
    </table>
    <audio
      id="audio"
      style="display: none"
      src="/static/audio/notification.mp3"
    ></audio>
    <script>
      const tbodyDOM = document.querySelector("#tbody");
      const audioDOM = document.querySelector("#audio");
      // 初始化 SSE 连接
      const eventSource = new EventSource("/subscribe/");

      eventSource.onmessage = function (event) {
        const data = JSON.parse(event.data);

        if (data.type === "new_order") {
          showNotification(
            `🔔 新订单提醒`,
            `客户：${data.customer}<br>金额：¥${data.amount}<br>单号：${data.order_id}`,
          );
          const tr = document.createElement("tr");
          for (let key in data) {
            if (key == "type") {
              continue;
            }
            const td = document.createElement("td");
            td.innerHTML = data[key];
            tr.append(td);
          }
          tbodyDOM.append(tr);
          // 可选：播放提示音
          // new Audio('/static/audio/notification.mp3').play();
          // audioDOM.play()
          const playButton = document.querySelector("#audio");
          const audio = new Audio("/static/audio/notification.mp3");

          playButton.addEventListener("click", () => {
            audio
              .play()
              .then(() => console.log("播放成功"))
              .catch((error) => console.error("播放失败:", error));
          });
        }
      };

      eventSource.onerror = function (err) {
        console.error("SSE 连接断开，尝试重连...", err);
        // EventSource 会自动重连，这里只需处理错误日志或 UI 提示
        // 如果连接彻底失败（如 403），可能需要手动关闭
      };

      // 显示通知 toast
      function showNotification(title, message) {
        const container = document.getElementById("notification-area");
        const toast = document.createElement("div");
        toast.className = "toast";
        toast.innerHTML = `<strong>${title}</strong><br>${message}`;
        container.appendChild(toast);
        // 5秒后移除
        setTimeout(() => {
          toast.style.opacity = "0";
          setTimeout(() => toast.remove(), 500);
        }, 5000);
      }
    </script>
  </body>
</html>

```

模拟用户端提交订单

```html
<!doctype html>
<html lang="en">
  <head>
    <meta charset="UTF-8" />
    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
    <title>提交订单</title>
    <style>
      * {
        padding: 0;
        margin: 0;
      }
      body {
        background-color: #f0f0f0;
      }
      .container {
        margin: 0 auto;
        width: 50%;
        background-color: #fff;
        padding: 20px;
      }
      .container h1{
        margin: 10px 0;
      }
      .container form{
        margin: 10px 0;
      }
      .container input[type='text']{
        padding: 6px;
        outline: none;
        width: 200px;
        border:1px solid #ccc;
        border-radius: 4px;
      }
    </style>
  </head>
  <body>
    <div class="container">
      <h1>提交订单</h1>
      <hr />
      <form action="/publish/" method="post">
        {% csrf_token %}
        <input
          required
          name="customer"
          type="text"
          placeholder="下单用户姓名"
        />
        <input required name="amount" type="text" placeholder="订单金额" />
        <button>提交订单</button>
      </form>
    </div>
  </body>
</html>

```

### 4. 路由配置 (urls.py)

```python
from django.urls import path
from sse import views
urlpatterns=[
    path('', views.index, name='index'),
    path('order_list/', views.order_list, name='order_list'),
    path('subscribe/', views.subscribe, name='subscribe'),
    path('publish/',views.Publish.as_view(),name="publish")
]
```

## 关键注意事项

1. **Nginx 配置 (至关重要)**
   如果你使用 Nginx 作为反向代理，必须关闭缓冲，否则消息会积压在 Nginx 直到缓冲区满才发给前端，导致“不实时”。

   ```nginx
   location /api/order-stream/ {
       proxy_pass http://127.0.0.1:8000;
       proxy_buffering off;      # 关闭缓冲
       proxy_cache off;          # 关闭缓存
       proxy_connect_timeout 60s;
       proxy_read_timeout 86400s; # 设置长超时，保持连接
       proxy_send_timeout 86400s;
   }
   ```

2. **并发与扩展性**

   - **单机版**：上面的 Redis 方案在单机或多机部署下都有效，因为 Redis 是中心化的消息 broker。无论 Django 开了几个进程/实例，只要连同一个 Redis，消息都能推送。

   - 海量连接

     ：如果后台管理员有上千人同时在线，每个连接占用一个线程/协程。Gunicorn (sync) 模式下可能会耗尽 worker。建议：

     - 使用 **Gevent** 或 **Uvicorn (Async)** 部署 Django。
     - 或者对于超大规模场景，考虑迁移到 **Django Channels** (基于 WebSocket/ASGI)，它对高并发长连接支持更好。但对于几十到几百人的后台系统，普通 Django + Redis + SSE 完全够用。

3. **权限控制**
   务必在 SSE 视图上加 `@login_required` 或自定义装饰器，确保只有登录的管理员才能建立连接，防止恶意用户占用服务器连接资源。

4. **消息丢失处理**
   SSE 在断开期间会丢失消息。如果需要保证“绝对不丢失”，可以在 reconnect 时利用 `Last-Event-ID` 机制，或者在前端重连后主动请求一次“最近未读消息”接口进行补偿。

## Redis Pub/Sub局限性

虽然 Redis Pub/Sub 很简单且快速，但在生产环境中使用时需注意以下几点：

1. **消息不持久化 (Fire-and-Forget)**：
   - 如果订阅者离线，发布的消息**会直接丢失**。Redis 不会为离线的客户端存储消息。
   - **解决方案**：如果需要消息持久化和历史回溯，请使用 **Redis Streams** (`XADD`, `XREAD`)。
2. **阻塞连接**：
   - 执行 `SUBSCRIBE` 或 `listen()` 后，该连接会被阻塞，只能用于接收消息，不能执行其他 Redis 命令（如 `GET`, `SET`）。通常需要为订阅单独开辟一个线程或连接。
3. **网络分区风险**：
   - 如果网络抖动导致订阅者断开重连，中间错过的消息无法补发。
4. **性能瓶颈**：
   - Pub/Sub 是单线程广播的。如果一个频道有数万个订阅者，或者消息体非常大，可能会阻塞 Redis 的主线程，影响其他命令的执行。
5. **无确认机制**：
   - 发布者不知道消息是否被成功处理，只知道有多少个订阅者收到了消息（TCP 层面送达）。

### **总结**

使用 **SSE + Redis Pub/Sub** 是实现后台订单提醒的**最佳实践**之一。它比轮询（每隔几秒请求一次数据库）效率高得多，比 WebSocket 实现起来简单得多，且完全满足实时性要求。