Django实时推送怎么实现

本文目录导读：

1. Django Channels（最推荐）
2. Server-Sent Events (SSE) + Django
3. WebSocket 第三方服务
4. 轮询 + Django（简单但效率低）
5. 最佳实践建议
6. 部署注意事项

在Django中实现实时推送，主要有以下几种方案,按推荐程度排序：

Django Channels（最推荐）

安装配置

pip install channels channels-redis

settings.py

INSTALLED_APPS = [
    'channels',
    # 其他应用
]
ASGI_APPLICATION = 'your_project.asgi.application'
CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {
            "hosts": [('127.0.0.1', 6379)],
        },
    },
}

consumers.py

import json
from channels.generic.websocket import AsyncWebsocketConsumer
class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.room_name = self.scope['url_route']['kwargs']['room_name']
        self.room_group_name = f'chat_{self.room_name}'
        # 加入房间
        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name
        )
        await self.accept()
    async def disconnect(self, close_code):
        # 离开房间
        await self.channel_layer.group_discard(
            self.room_group_name,
            self.channel_name
        )
    async def receive(self, text_data):
        text_data_json = json.loads(text_data)
        message = text_data_json['message']
        # 发送到房间
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                'type': 'chat_message',
                'message': message
            }
        )
    async def chat_message(self, event):
        message = event['message']
        await self.send(text_data=json.dumps({
            'message': message
        }))

routing.py

from django.urls import re_path
from . import consumers
websocket_urlpatterns = [
    re_path(r'ws/chat/(?P<room_name>\w+)/$', consumers.ChatConsumer.as_asgi()),
]

asgi.py

import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
import your_app.routing
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')
application = ProtocolTypeRouter({
    'http': get_asgi_application(),
    'websocket': AuthMiddlewareStack(
        URLRouter(
            your_app.routing.websocket_urlpatterns
        )
    ),
})

前端JavaScript

const chatSocket = new WebSocket(
    'ws://' + window.location.host + '/ws/chat/' + roomName + '/'
);
chatSocket.onmessage = function(e) {
    const data = JSON.parse(e.data);
    // 处理接收到的消息
};
chatSocket.onclose = function(e) {
    console.error('Chat socket closed unexpectedly');
};
// 发送消息
document.querySelector('#chat-message-input').onkeyup = function(e) {
    if (e.keyCode === 13) {
        chatSocket.send(JSON.stringify({
            'message': this.value
        }));
    }
};

Server-Sent Events (SSE) + Django

视图

from django.http import StreamingHttpResponse
import time
import json
def event_stream(request):
    def event_generator():
        while True:
            # 模拟实时数据
            data = {
                'message': f'Current time: {time.time()}',
                'timestamp': time.time()
            }
            yield f"data: {json.dumps(data)}\n\n"
            time.sleep(1)
    response = StreamingHttpResponse(
        event_generator(),
        content_type='text/event-stream'
    )
    response['Cache-Control'] = 'no-cache'
    response['X-Accel-Buffering'] = 'no'
    return response

前端

const eventSource = new EventSource('/stream/');
eventSource.onmessage = function(e) {
    const data = JSON.parse(e.data);
    console.log('Received:', data);
};
eventSource.onerror = function(e) {
    console.error('EventSource failed');
};

WebSocket 第三方服务

使用 Pusher（第三方服务）

# pip install pusher
import pusher
pusher_client = pusher.Pusher(
    app_id='your-app-id',
    key='your-key',
    secret='your-secret',
    cluster='ap1'
)
# 触发事件
pusher_client.trigger('my-channel', 'my-event', {'message': 'hello world'})

使用 Firebase Cloud Messaging (FCM)

# pip install firebase-admin
import firebase_admin
from firebase_admin import credentials, messaging
cred = credentials.Certificate('serviceAccountKey.json')
firebase_admin.initialize_app(cred)
def send_push_notification(registration_token, title, body):
    message = messaging.Message(
        notification=messaging.Notification(
            title=title,
            body=body,
        ),
        token=registration_token,
    )
    response = messaging.send(message)

轮询 + Django（简单但效率低）

视图

from django.http import JsonResponse
from django.views.decorators.cache import never_cache
@never_cache
def poll_view(request):
    # 查询最新数据
    latest_data = get_latest_data()
    return JsonResponse(latest_data)

前端

function pollServer() {
    fetch('/poll/')
        .then(response => response.json())
        .then(data => {
            console.log('Received:', data);
        });
}
// 每2秒轮询一次
setInterval(pollServer, 2000);

最佳实践建议

1. 推荐使用 Channels：如果项目需要复杂实时功能
2. 中小项目用 SSE：简单，浏览器原生支持
3. 生产环境：配合 Redis、Nginx 反向代理
4. 移动端：考虑 WebSocket 或第三方推送服务
5. 性能优化：
   • 使用 Redis 作为 Channel Layer
   • 配置 Celery 处理异步任务
   • 使用异步视图（Django 3.1+）

部署注意事项

# Nginx 配置示例
location /ws/ {
    proxy_pass http://daphne:8001;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
    proxy_set_header Host $host;
}

选择方案时，考虑项目规模、实时性要求、团队技术栈和部署环境，Channels 是最完整的方案，但学习曲线稍陡；SSE 简单但功能有限；第三方服务方便但可能产生费用。