django StreamingHttpResponse fetchEventSource实现前后端流试返回数据并接收数据的完整详细过程

news2025/1/4 5:00:25

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,也给注释了则在切换页面后会重新接收数据，效果如下

在这里插入图片描述
跨域设置可参考
流试说明可参考

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/2269259.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

如何在 Ubuntu 22.04 上安装 Webmin 教程

如何在 Ubuntu 22.04 上安装 Webmin 教程

简介 在本教程中&#xff0c;我们将解释如何在 Ubuntu 22.04 服务器上安装 Webmin。 Webmin 是一个功能强大的基于 Web 的控制面板&#xff0c;它允许你通过一个简单的 Web 界面管理服务器的各个方面&#xff0c;例如用户帐户、DNS、防火墙、数据库等等。本指南将引导你完成在…
阅读更多...
【一起python】银行管理系统

【一起python】银行管理系统

文章目录 &#x1f4dd;计算机基础概念&#x1f320; 导入模块&#x1f320;定义input_card_info函数&#x1f320; 定义check_password函数&#x1f320;初始化用户字典和欢迎信息&#x1f309; 主循环&#x1f309;开户操作&#x1f309;查询操作&#x1f309;取款操作&#…
阅读更多...
【D3.js in Action 3 精译_047】5.2：图形的堆叠（一）—— 图解 D3 中的堆叠布局生成器

【D3.js in Action 3 精译_047】5.2：图形的堆叠（一）—— 图解 D3 中的堆叠布局生成器

当前内容所在位置&#xff1a; 第五章 饼图布局与堆叠布局 ✔️ 5.1 饼图和环形图的创建 5.1.1 准备阶段&#xff08;一&#xff09;5.1.2 饼图布局生成器&#xff08;二&#xff09;5.1.3 圆弧的绘制&#xff08;三&#xff09;5.1.4 数据标签的添加&#xff08;四&#xff09…
阅读更多...
自建私有云相册：Docker一键部署Immich，照片视频备份利器

自建私有云相册：Docker一键部署Immich，照片视频备份利器

自建私有云相册&#xff1a;Docker一键部署Immich&#xff0c;照片视频备份利器 前言 随着人们手机、PC、平板等电子产品多样&#xff0c;我们拍摄和保存的照片和视频数量也在不断增加。如何高效地管理和备份这些珍贵的记忆成为了一个重要的问题。 传统的云备份虽然方便&…
阅读更多...
[微服务] - MQ高级

[微服务] - MQ高级

在昨天的练习作业中&#xff0c;我们改造了余额支付功能&#xff0c;在支付成功后利用RabbitMQ通知交易服务&#xff0c;更新业务订单状态为已支付。 但是大家思考一下&#xff0c;如果这里MQ通知失败&#xff0c;支付服务中支付流水显示支付成功&#xff0c;而交易服务中的订单…
阅读更多...
【Unity3D】A*寻路（2D究极简单版）

【Unity3D】A*寻路（2D究极简单版）

运行后点击透明格子empty即执行从(0,0)起点到点击为止终点&#xff08;测试是(5,5)&#xff09;如下图 UICamera深度要比MainCamera大&#xff0c;Clear Flags:Depth only&#xff0c;正交视野 MainCamera保持原样&#xff1b;注意Line绘线物体的位置大小旋转信息&#xff0c;不…
阅读更多...
xadmin后台首页增加一个导入数据按钮

xadmin后台首页增加一个导入数据按钮

xadmin后台首页增加一个导入数据按钮 效果 流程 1、在添加小组件中添加一个html页面 2、写入html代码 3、在urls.py添加导入数据路由 4、在views.py中添加响应函数html代码 <!DOCTYPE html> <html lang
阅读更多...
压敏电阻MOV选型【EMC】

压敏电阻MOV选型【EMC】

左侧的压敏电阻用来防护差模干扰&#xff1b;右侧并联在L N 两端的压敏电阻是用来防护共模干扰&#xff1a; 选择压敏电阻时&#xff0c;通常需要考虑以下几个关键因素&#xff0c;以确保它能够有效保护电路免受浪涌电流或过电压的损害&#xff0c;同时满足 EMC 要求&#xff1…
阅读更多...
pycharm pytorch tensor张量可视化，view as array

pycharm pytorch tensor张量可视化，view as array

Evaluate Expression 调试过程中&#xff0c;需要查看比如attn_weight 张量tensor的值。 方法一&#xff1a;attn_weight.detach().numpy(),view as array 方法二&#xff1a;attn_weight.cpu().numpy(),view as array
阅读更多...
log4j2的Strategy、log4j2的DefaultRolloverStrategy、删除过期文件

log4j2的Strategy、log4j2的DefaultRolloverStrategy、删除过期文件

文章目录 一、DefaultRolloverStrategy1.1、DefaultRolloverStrategy节点1.1.1、filePattern属性1.1.2、DefaultRolloverStrategy删除原理 1.2、Delete节点1.2.1、maxDepth属性 二、知识扩展2.1、DefaultRolloverStrategy与Delete会冲突吗&#xff1f;2.1.1、场景一&#xff1a…
阅读更多...
设计模式之访问者模式：一楼千面 各有玄机

设计模式之访问者模式：一楼千面 各有玄机

~犬&#x1f4f0;余~ “我欲贱而贵&#xff0c;愚而智&#xff0c;贫而富&#xff0c;可乎&#xff1f; 曰&#xff1a;其唯学乎” 一、访问者模式概述 \quad 江湖中有一个传说&#xff1a;在遥远的东方&#xff0c;有一座神秘的玉楼。每当武林中人来访&#xff0c;楼中的各个房…
阅读更多...
结合实例来聊聊UDS诊断中的0x2F服务

结合实例来聊聊UDS诊断中的0x2F服务

1、什么是UDS中的0x2F服务 0x2F简单来说&#xff0c;就是输入输出控制服务。先看官方的简绍 翻译如下&#xff1a; InputOutputControlByldentifier服务来替换输入信号、内部服务器函数和/或强制控制为电子系统的输出&#xff08;执行器&#xff09;的值。通常&#xff0c;此…
阅读更多...
1月第二讲：WxPython跨平台开发框架之图标选择界面

1月第二讲：WxPython跨平台开发框架之图标选择界面

1、图标分类介绍 这里图标我们分为两类&#xff0c;一类是wxPython内置的图标资源&#xff0c;以wx.Art_开始。wx.ART_ 是 wxPython 提供的艺术资源&#xff08;Art Resource&#xff09;常量&#xff0c;用于在界面中快速访问通用的图标或位图资源。这些资源可以通过 wx.ArtP…
阅读更多...
【弱监督视频异常检测】2024-TCSVT-基于片段间特征相似度的多尺度时间 MLP 弱监督视频异常检测

【弱监督视频异常检测】2024-TCSVT-基于片段间特征相似度的多尺度时间 MLP 弱监督视频异常检测

2024-TCSVT-Inter-clip Feature Similarity based Weakly Supervised Video Anomaly Detection via Multi-scale Temporal MLP 基于片段间特征相似度的多尺度时间 MLP 弱监督视频异常检测摘要1. 引言2. 相关工作A. 分布外检测B. 弱监督视频异常检测C. 多层感知器 3. 方法A. 概述…
阅读更多...
C# OpenCV机器视觉:凸包检测

C# OpenCV机器视觉:凸包检测

在一个看似平常却又暗藏玄机的午后&#xff0c;阿强正悠闲地坐在实验室里&#xff0c;翘着二郎腿&#xff0c;哼着小曲儿&#xff0c;美滋滋地品尝着手中那杯热气腾腾的咖啡&#xff0c;仿佛整个世界都与他无关。突然&#xff0c;实验室的门 “砰” 的一声被撞开&#xff0c;小…
阅读更多...
【英特尔IA-32架构软件开发者开发手册第3卷：系统编程指南】2001年版翻译，2-44

【英特尔IA-32架构软件开发者开发手册第3卷：系统编程指南】2001年版翻译，2-44

文件下载与邀请翻译者 学习英特尔开发手册&#xff0c;最好手里这个手册文件。原版是PDF文件。点击下方链接了解下载方法。 讲解下载英特尔开发手册的文章 翻译英特尔开发手册&#xff0c;会是一件耗时费力的工作。如果有愿意和我一起来做这件事的&#xff0c;那么&#xff…
阅读更多...
8.若依系统监控与定时任务

8.若依系统监控与定时任务

帮助开发者和运维快速了解应用程序的性能状态。 数据监控 定时任务 实现动态管理任务。 需求&#xff1a;每间隔5s&#xff0c;控制台输出系统时间。 新建的任务类必须在指定目录ruoyi-quartz模块下的task包下。 状态设置为启动 执行策略 场景&#xff1a;比如一个任务每个…
阅读更多...
【JAVA高级篇教学】第六篇：Springboot实现WebSocket

【JAVA高级篇教学】第六篇：Springboot实现WebSocket

在 Spring Boot 中对接 WebSocket 是一个常见的场景&#xff0c;通常用于实现实时通信。以下是一个完整的 WebSocket 集成步骤&#xff0c;包括服务端和客户端的实现。本期做个简单的测试用例。 目录 一、WebSocket 简介 1. 什么是 WebSocket&#xff1f; 2. WebSocket 的特…
阅读更多...
【YOLO 项目实战】（12）红外/可见光多模态目标检测

【YOLO 项目实战】（12）红外/可见光多模态目标检测

欢迎关注『youcans动手学模型』系列 本专栏内容和资源同步到 GitHub/youcans 【YOLO 项目实战】&#xff08;10&#xff09;YOLO8 环境配置与推理检测 【YOLO 项目实战】&#xff08;11&#xff09;YOLO8 数据集与模型训练 【YOLO 项目实战】&#xff08;12&#xff09;红外/可…
阅读更多...
Ubuntu开机The root filesystem on /dev/sdbx requires a manual fsck 问题

Ubuntu开机The root filesystem on /dev/sdbx requires a manual fsck 问题

出现“Manual fsck”错误可能由以下几种原因引起&#xff1a; 不正常关机&#xff1a;如果系统意外断电或被强制重启&#xff0c;文件系统可能未能正确卸载&#xff0c;导致文件系统损坏。磁盘故障&#xff1a;硬盘的物理损坏可能会引发文件系统错误。文件系统配置问题&#x…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023