django后端环境介绍：
Python 3.10.14
pip install django-cors-headers==4.4.0 Django==5.0.6 django-cors-headers==4.4.0 djangorestframework==3.15.2 -i https://mirrors.tuna.tsinghua.edu.cn/pypi/web/simple
总环境如下：
Package Version
------------------- -------
asgiref 3.8.1
Django 5.0.6
django-cors-headers 4.4.0
djangorestframework 3.15.2
pip 24.2
setuptools 75.1.0
sqlparse 0.5.3
typing_extensions 4.12.2
wheel 0.44.0
1、创建项目
django-admin startproject event_stream_test
cd event_stream_test
python manage.py startapp sse
总的目录结构如下：
event_stream_test/settings.py中的配置如下：
INSTALLED_APPS = [
......
"sse.apps.SseConfig",
"corsheaders",
......
]
MIDDLEWARE = [
......
"django.contrib.sessions.middleware.SessionMiddleware",
"corsheaders.middleware.CorsMiddleware", # 这个位置要在CommonMiddleware之前
"django.middleware.common.CommonMiddleware",
......
]
# 允许跨域设置
CORS_ALLOW_CREDENTIALS = True
CORS_ALLOW_ALL_ORIGINS = True
CORS_ALLOW_HEADERS = ('*')
完整的settings设置如下：
"""
Django settings for event_stream_test project.
Generated by 'django-admin startproject' using Django 4.2.
For more information on this file, see
https://docs.djangoproject.com/en/4.2/topics/settings/
For the full list of settings and their values, see
https://docs.djangoproject.com/en/4.2/ref/settings/
"""
import os
from pathlib import Path
# Build paths inside the project like this: BASE_DIR / 'subdir'.
BASE_DIR = Path(__file__).resolve().parent.parent
# Quick-start development settings - unsuitable for production
# See https://docs.djangoproject.com/en/4.2/howto/deployment/checklist/
# SECURITY WARNING: keep the secret key used in production secret!
SECRET_KEY = "django-insecure-u@my#&53d3%e79ox#u!rs0eprc$c)_2gbcezpdj0v^=_(!wwu-"
# SECURITY WARNING: don't run with debug turned on in production!
DEBUG = True
ALLOWED_HOSTS = []
# Application definition
INSTALLED_APPS = [
"django.contrib.admin",
"django.contrib.auth",
"django.contrib.contenttypes",
"django.contrib.sessions",
"django.contrib.messages",
"django.contrib.staticfiles",
"sse.apps.SseConfig",
"corsheaders",
# "rest_framework"
]
MIDDLEWARE = [
"django.middleware.security.SecurityMiddleware",
"django.contrib.sessions.middleware.SessionMiddleware",
"corsheaders.middleware.CorsMiddleware",
"django.middleware.common.CommonMiddleware",
"django.middleware.csrf.CsrfViewMiddleware",
"django.contrib.auth.middleware.AuthenticationMiddleware",
"django.contrib.messages.middleware.MessageMiddleware",
"django.middleware.clickjacking.XFrameOptionsMiddleware",
]
ROOT_URLCONF = "event_stream_test.urls"
TEMPLATES = [
{
"BACKEND": "django.template.backends.django.DjangoTemplates",
# "DIRS": [os.path.join(BASE_DIR, 'sse/templates')],
"DIRS": [],
"APP_DIRS": True,
"OPTIONS": {
"context_processors": [
"django.template.context_processors.debug",
"django.template.context_processors.request",
"django.contrib.auth.context_processors.auth",
"django.contrib.messages.context_processors.messages",
],
},
},
]
WSGI_APPLICATION = "event_stream_test.wsgi.application"
# Database
# https://docs.djangoproject.com/en/4.2/ref/settings/#databases
DATABASES = {
"default": {
"ENGINE": "django.db.backends.sqlite3",
"NAME": BASE_DIR / "db.sqlite3",
}
}
# Password validation
# https://docs.djangoproject.com/en/4.2/ref/settings/#auth-password-validators
AUTH_PASSWORD_VALIDATORS = [
{
"NAME": "django.contrib.auth.password_validation.UserAttributeSimilarityValidator",
},
{
"NAME": "django.contrib.auth.password_validation.MinimumLengthValidator",
},
{
"NAME": "django.contrib.auth.password_validation.CommonPasswordValidator",
},
{
"NAME": "django.contrib.auth.password_validation.NumericPasswordValidator",
},
]
# Internationalization
# https://docs.djangoproject.com/en/4.2/topics/i18n/
LANGUAGE_CODE = "en-us"
TIME_ZONE = "UTC"
USE_I18N = True
USE_TZ = True
# Static files (CSS, JavaScript, Images)
# https://docs.djangoproject.com/en/4.2/howto/static-files/
STATIC_URL = "static/"
# Default primary key field type
# https://docs.djangoproject.com/en/4.2/ref/settings/#default-auto-field
DEFAULT_AUTO_FIELD = "django.db.models.BigAutoField"
CORS_ALLOW_CREDENTIALS = True
CORS_ALLOW_ALL_ORIGINS = True
CORS_ALLOW_HEADERS = ('*')
event_stream_test/urls.py中的配置如下：
from django.contrib import admin
from django.urls import path, include
urlpatterns = [
path("admin/", admin.site.urls),
path('v1/sse/', include('sse.urls'))
]
sse/urls.py中的配置如下：
from django.urls import path
from sse import views
urlpatterns = [
# SSE
path('demo', views.DemoAPIView.as_view()),
]
sse/views.py中的配置如下：
import time
from django.http import StreamingHttpResponse
from rest_framework.views import APIView
class DemoAPIView(APIView):
# def post(self, request):
def get(self, request):
# 定义一个生成器，持续输出数据流
def event_stream():
# 测试读取当前文件
# with open('sse/1.txt', mode="r", encoding='utf-8') as file:
# for line in file:
# time.sleep(3)
# yield f'data: {line}\n\n'
#
count = 0
while True:
count = count+1
line = "当前行号是"+str(count)
yield f'data: {line}\n\n'
time.sleep(3)
response = StreamingHttpResponse(event_stream(), content_type='text/event-stream; charset=utf-8',
headers={'Cache-Control': 'no-cache'}, status=200)
return response
至此配置django配置已经完成，运行项目则可以直接在浏览器中看到结果：
python manage.py runserver
然后在浏览器中输入：
http://127.0.0.1:8000/v1/sse/demo
即可看到如下内容，会一直不停地打印
前端环境介绍：
"dependencies": {
"@microsoft/fetch-event-source": "^2.0.1",
"vue": "^3.5.13"
},
需要安装npm i @microsoft/fetch-event-source
最好版本一致
Vue3代码如下（App.vue下）
<template>
<h1>这是测试页</h1>
<h1 v-for="item in data" style="color: red">
{{ item }}
</h1>
</template>
<script setup>
import {ref, onMounted, onUnmounted} from "vue";
import {fetchEventSource} from '@microsoft/fetch-event-source';
const controller = new AbortController();
const data = ref([])
const connectSSE = async () => {
const url = "http://127.0.0.1:8000/v1/sse/demo"
console.log(url)
await fetchEventSource(url, {
method: 'GET',
headers: {
'Accept': '*/*', // 这个很重要
// "Content-Type": "text/event-stream"
},
// 发送请求的内容可以放在body中
// body: JSON.stringify({
// "type": "auto",
// "text": "阿斯蒂芬",
// "background": "",
// "messages": [],
// "meta_data": {}
// }),
signal: controller.signal,
openWhenHidden: true, // 当不在当前页面时页面依然能不间断的接收数据
onmessage: async (event) => {
console.log('event', event)
// console.log("JSON.parse(event.data)===》", JSON.parse(event.data))
// console.log('event.data.split("是")', event.data.split("是"));
let num = parseInt(event.data.split("是")[1])
console.log("num", num)
if (num === 3){
console.log("终止进程没") // 在上边写这个signal: controller.signal,在这写controller.abort()可以终止接收后端的推送
controller.abort()
}else {
data.value.push(event.data)
}
},
onerror(err) {
console.error('Error:', err);
if (err.status === 500) {
// 服务器错误时重新连接
setTimeout(() => connectSSE(), 5000);
}
},
onopen(response) {
if (response.ok) {
console.log('Connection start');
}
},
onclose() {
console.log('Connection close');
}
})
}
onMounted(() => {
connectSSE();
});
onUnmounted(() => {
});
</script>
运行前端项目npm run dev 并打开前端页面，展示结果如下：
如果把前端中的这段代码替换成data.value.push(event.data)
//if (num === 3){
// console.log("终止进程没") // 在上边写这个signal: controller.signal,在这写controller.abort()可以终止接收后端的推送
// controller.abort()
//}else {
// data.value.push(event.data)
//}
替换为：
data.value.push(event.data)
则效果如下：会一直打印，不会终止接收，所以controller.abort()得作用就是终止接收后端的推送
如果把openWhenHidden: true,也给注释了则在切换页面后会重新接收数据，效果如下
跨域设置可参考
流试说明可参考
