本文目录导读：
在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);
最佳实践建议
- 推荐使用 Channels：如果项目需要复杂实时功能
- 中小项目用 SSE：简单，浏览器原生支持
- 生产环境：配合 Redis、Nginx 反向代理
- 移动端：考虑 WebSocket 或第三方推送服务
- 性能优化：
- 使用 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 简单但功能有限；第三方服务方便但可能产生费用。
