老实说，Django本身并不支持它，但我有一个使用Daphne的解决方案（它也在Django通道中使用）。
创建了自己的StreamingHttpResponse类，能够从异步方法中检索数据流，并将其提供给Django的同步部分。

import asyncio

# By design asyncio does not allow its event loop to be nested.
# Trying to do so will give the error "RuntimeError: This event loop is already running".
# This library solves that problem.
import nest_asyncio

from django.http.response import StreamingHttpResponse

class AsyncStreamingHttpResponse(StreamingHttpResponse):

    def __init__(self, streaming_content=(), *args, **kwargs):
        sync_streaming_content = self.get_sync_iterator(streaming_content)
        super().__init__(streaming_content=sync_streaming_content, *args, **kwargs)

    @staticmethod
    async def convert_async_iterable(stream):
        """Accepts async_generator and async_iterator"""
        return iter([chunk async for chunk in stream])

    def get_sync_iterator(self, async_iterable):
        nest_asyncio.apply()

        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        result = loop.run_until_complete(self.convert_async_iterable(async_iterable))
        return result

此外，您需要使用Daphne运行Django Web服务器以正确支持服务器发送事件（SSE）。它由“Django Software Foundation”正式支持，并且具有与gunicorn类似的语法，但使用asgi.py而不是wsgi.py。
要使用它-您可以使用以下命令安装：pip install daphne
并将命令更改为：python manage.py runserver
比如说daphne -b 0.0.0.0 -p 8000 sse_demo.asgi:application。
不确定它是否能与gunicorn一起工作。
如果你还有问题就告诉我。

另一种执行SSE的方法是使用特殊的库django-eventstream：
将以下内容添加到将使用数据的HTML页面：

<script src="{% static 'django_eventstream/eventsource.min.js' %}"></script>
<script src="{% static 'django_eventstream/reconnecting-eventsource.js' %}"></script>

var es = new ReconnectingEventSource('/events/');

es.addEventListener('message', function (e) {
    console.log(e.data);
}, false);

es.addEventListener('stream-reset', function (e) {
    // ... client fell behind, reinitialize ...
}, false);

对于后端，你需要正确地设置Django，然后你就可以在任何需要做服务器端事件（SSE）的视图/任务/信号/方法中调用以下方法：
添加以下将生成数据（事件）的视图：

# from django_eventstream import send_event

send_event('test', 'message', {'text': 'hello world'})

我创建了一个名为stream的装饰器，它可以与一个协程函数一起使用，使其与Django的StreamingHttpResponse兼容。下面是一个例子：

import asyncio
import functools

from django.http import StreamingHttpResponse

def stream(coroutine_function):
    @functools.wraps(coroutine_function)
    def wrapper(*args, **kwargs):
        coroutine = coroutine_function(*args, **kwargs)
        try:
            while True:
                yield asyncio.run(coroutine.__anext__())
        except StopAsyncIteration:
            pass
    return wrapper

@stream
async def chunks():
    for char in 'Hello, world!':
        yield char
        await asyncio.sleep(1)

async def index(request):
    return StreamingHttpResponse(chunks())

我还需要添加nest_asyncio并在settings.py文件的顶部调用apply()，如下所示：

import nest_asyncio
nest_asyncio.apply()

nest_asyncio依赖项支持从stream装饰器创建的wrapper函数调用asyncio.run。
最后，Django的asgi可以使用uvicorn到gunicorn运行，如下所示：

$ gunicorn -k uvicorn.workers.UvicornWorker www.asgi:application

你可以使用django-channel：
Channels增强了Django，为您的代码带来了WebSocket，long-poll HTTP，任务卸载和其他异步支持，使用熟悉的Django设计模式和灵活的底层框架，不仅可以自定义行为，还可以为您自己的协议和需求编写支持。