如何在Django中实现SSE

Open Table of contents

概述
设计
- Daphne
- Django APP

概述

查看数据状态是日常Web开发中经常会遇到的一种场景，比较常见的解决方案是轮询（Polling）。它是最简单的实现方式即每隔一段时间想服务器发送请求，然后服务器把状态返回给客户端。 implement-sse-1 但是这种方案的不足之处在于，它会浪费带宽以及服务器的响应。因为大多数情况下，如果数据没有更新的话，服务器返回的是重复的内容。

为了优化轮询，Long-Polling（长轮询）和Server-Sent Event是常见的方式。前者在服务器接收到客户端请求后，不立即返回，而是保持连接状态直到它有数据更新后才返回。后者则是单向通信，从服务器返回数据，客户端只负责接收，不能发送数据。这两种方式都很好的减少了不必要的负载以及返回有效的数据。 implement-sse-2

今天，我们只讨论SSE，如果有兴趣想了解长轮询的朋友，可以通过下面的联系方式找到我，我会倾囊相授。 implement-sse-3

设计

前端用HTMX的扩展件sse.js来发起请求，当Django收到请求后利用asyncio.sleep模拟服务器进行长时间的运算,每隔一段时间返回一个随机emoji。并且我们会用Daphne作为异步服务器，这样可以让Django在后台执行长时间的运算时不会阻塞新的请求。

Daphne

配置Daphne非常简单！这边我会默认我们已经安装了Django并且激活了虚拟环境。

安装 pip install daphne
修改settings.py

INSTALLED_APPS = [
    "daphne",
    ...,
]

ASGI_APPLICATION = "myproject.asgi.application"
# 这里的myproject是我们的Django项目，不是Django APP哦！
# 如果不确定项目名字叫什么，找下asgi.py和wsgi.py所在的文件夹，该文件夹名字就是项目名字。

启动runserver python .\manage.py runserver

可以看到，web server已被替换成ASGI/Daphne。如果我们把INSTALLED_APPS里面的daphne注释掉，可以回到默认的开发服务器。 implement-sse-4

当然，这里讨论的只是开发环境，生产环境可以参考Django官方文档。

Django APP

我们创建一个新的APP来实现SSE的功能：

python .\manage.py startapp sse

在settings.py的INSTALLED_APPS中添加新的APP:

INSTALLED_APPS = [
    "daphne",
    ...,
    "sse",
]

在APP目录中，新建一个urls.py文件来处理指向这个应用的请求，同时在项目的urls.py(myproject/urls.py)中用关键字include重定向所有“stream/*”的请求

# myproject/urls.py
from django.contrib import admin
from django.urls import path,include

urlpatterns = [
    path("stream/", include("sse.urls")),
    path('admin/', admin.site.urls),
]

#------------------------------------------------#

# sse/urls.py
from django.urls import path
from . import views

urlpatterns = [
    path("", views.sse_stream, name="sse_stream"),
]

当用户请求“https://127.0.0.1:8000/stream/”的时候，视图函数sse_stream会被调用。

视图函数sse_stream：

import asyncio
import random
from django.http import StreamingHttpResponse

async def event_stream():
    emojis = ["🍔","🎈","🎃","🧦"]
    i = 0
    while True:
        yield f'data: {random.choice(emojis)} {i}\n\n'
        i += 1
        await asyncio.sleep(3)

async def sse_stream(request):
    return StreamingHttpResponse(event_stream(), content_type='text/event-stream')

虽然StreamingHttpResponse也可以在WSGI服务器（同步）上返回，但是相应的event_stream()必须返回同步迭代器，得删除关键字async和await，并且把asyncio替换成time。这里我们稍微跑题了一下，但我觉得还是有必要了解Django对于同步和异步响应的不同处理。

请注意，content_type必须是text/event-stream，让客户端知道接收的响应是SSE，详情可参考MDN。

因为我们用异步实现SSE，所以生成器event_stream()也得是异步的。如果我们需要处理长时间的任务，那么可以写一个异步函数替换asyncio.sleep(),并通过yield返回状态，这里就不展开了。

前端: 这里我们先得引用HTMX的SSE插件，可以用CDN或者下载后放在static文件夹下面。

<script src="https://unpkg.com/htmx.org/dist/ext/sse.js"></script>
!<-- or -->
<script src="{% static 'form/js/sse.js' %}"></script>

接着，按照使用方法设置请求URL（sse-connect）和监听事件（sse-swap）

<div hx-ext="sse" sse-connect="{% url 'sse_stream' %}" sse-swap="message">
  Emoji starts from here!
</div>

hx-ext: 专门用来调用插件的，可以参考这个插件列表
sse-connect: SSE请求的路径“stream/”
sse-swap: 监听SSE事件的类型。按照SSE的定义，在响应中可以包含“event”，“data”， “id”，“retry”这四个属性，当没有指定event的话，那么就默认用“message”作为它的值。而我们的视图函数里面只用到了“data”，没有“event”,所以监听的事件就对应为“message”。

一旦加载了该页面，那么HTMX就会立即发送请求到“stream/”，这对于某些场景下不是非常友好，比如“点击某个按钮之后再触发SSE”。所以我们可以利用hx-get的特性，当用户点击时去请求上面的代码块<div hx-ext="sse" ...></div>，当该代码块做为响应返回并加载到页面后才触发SSE。

好了，运行代码，打开DevTools，切换到Network就可以看到SSE事件，它的类型（event）和数据（data）一览无余。 implement-sse-5