文章目录
- 准备工作
- 创建 Django 项目
- 创建应用程序
- 配置项目
- 编写 Consumer
- 编写路由
- 创建 URL 路由
- 运行应用
- 用户认证
- 消息持久化
- 显示历史消息
- 结论
Django Channels 是 Django 的一个扩展,允许在 Web 应用中添加实时功能,例如 Websockets、HTTP2 和其他协议。本文将介绍如何使用 Django Channels 构建一个简单的实时聊天应用程序。
准备工作
首先,确保你已经安装了 Django 和 Channels。你可以使用以下命令安装:
pip install django channels
创建 Django 项目
使用以下命令创建一个新的 Django 项目:
django-admin startproject realtime_chat
然后进入项目目录:
cd realtime_chat
创建应用程序
创建一个新的 Django 应用程序:
python manage.py startapp chat
配置项目
在 settings.py
文件中添加 Channels 的配置:
# settings.py
INSTALLED_APPS = [
...
'channels',
'chat',
]
ASGI_APPLICATION = 'realtime_chat.routing.application'
创建一个新的 ASGI 路由文件 routing.py
:
# routing.py
from channels.routing import ProtocolTypeRouter
application = ProtocolTypeRouter({
})
编写 Consumer
在 consumers.py
文件中编写一个简单的 Consumer:
# chat/consumers.py
from channels.generic.websocket import AsyncWebsocketConsumer
class ChatConsumer(AsyncWebsocketConsumer):
async def connect(self):
await self.accept()
async def disconnect(self, close_code):
pass
async def receive(self, text_data):
pass
编写路由
在 routing.py
文件中添加路由配置:
# routing.py
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from django.urls import path
from chat.consumers import ChatConsumer
application = ProtocolTypeRouter({
'websocket': AuthMiddlewareStack(
URLRouter([
path('chat/', ChatConsumer.as_asgi()),
])
),
})
创建 URL 路由
在 urls.py
文件中添加 URL 路由:
# chat/urls.py
from django.urls import path
from . import consumers
websocket_urlpatterns = [
path('chat/', consumers.ChatConsumer.as_asgi()),
]
在主项目的 urls.py
中包含 Chat 应用的路由:
# realtime_chat/urls.py
from django.urls import path, include
urlpatterns = [
...
path('chat/', include('chat.urls')),
]
运行应用
运行 Django 服务器:
python manage.py runserver
现在,你可以通过访问 http://localhost:8000/chat/
来测试你的实时聊天应用了。
用户认证
首先,我们将使用 Django 自带的认证系统来处理用户认证。在 settings.py
中启用认证系统:
# settings.py
INSTALLED_APPS = [
...
'django.contrib.auth',
'django.contrib.contenttypes',
...
]
AUTHENTICATION_BACKENDS = [
'django.contrib.auth.backends.ModelBackend',
]
然后,我们需要修改 ChatConsumer
,以便处理用户认证:
# chat/consumers.py
from channels.generic.websocket import AsyncWebsocketConsumer
from django.contrib.auth.models import AnonymousUser
class ChatConsumer(AsyncWebsocketConsumer):
async def connect(self):
self.user = self.scope["user"]
if isinstance(self.user, AnonymousUser):
await self.close()
else:
await self.accept()
async def disconnect(self, close_code):
pass
async def receive(self, text_data):
pass
现在,只有经过身份验证的用户才能连接到聊天消费者。
消息持久化
我们希望用户在重新加载页面后能够看到之前的聊天消息。为了实现这一点,我们将使用 Django 的数据库来存储消息。
首先,我们需要创建一个模型来存储聊天消息:
# chat/models.py
from django.db import models
from django.contrib.auth.models import User
class Message(models.Model):
user = models.ForeignKey(User, on_delete=models.CASCADE)
content = models.TextField()
timestamp = models.DateTimeField(auto_now_add=True)
def __str__(self):
return f"{self.user.username}: {self.content}"
然后,我们需要修改 ChatConsumer
,以便在接收到消息时将其保存到数据库中:
# chat/consumers.py
from channels.generic.websocket import AsyncWebsocketConsumer
from django.contrib.auth.models import AnonymousUser
from .models import Message
import json
class ChatConsumer(AsyncWebsocketConsumer):
async def connect(self):
self.user = self.scope["user"]
if isinstance(self.user, AnonymousUser):
await self.close()
else:
await self.accept()
async def disconnect(self, close_code):
pass
async def receive(self, text_data):
text_data_json = json.loads(text_data)
content = text_data_json['message']
message = Message.objects.create(user=self.user, content=content)
message.save()
现在,当用户发送消息时,它们将被保存到数据库中。
显示历史消息
最后,我们需要修改前端以显示历史消息。在 ChatConsumer
中,我们可以添加一个方法来发送历史消息给客户端:
# chat/consumers.py
class ChatConsumer(AsyncWebsocketConsumer):
...
async def fetch_messages(self):
messages = Message.objects.all()[:10] # 获取最近的10条消息
content = {
'command': 'messages',
'messages': self.messages_to_json(messages)
}
await self.send(text_data=json.dumps(content))
@staticmethod
def messages_to_json(messages):
result = []
for message in messages:
result.append({
'user': message.user.username,
'content': message.content,
'timestamp': str(message.timestamp)
})
return result
然后,在连接建立时调用这个方法:
# chat/consumers.py
async def connect(self):
...
await self.fetch_messages()
在前端,你可以使用 JavaScript 来接收并显示历史消息。
结论
通过添加用户认证和消息持久化功能,我们的实时聊天应用变得更加完善和实用。你可以根据需要进一步定制和扩展这个应用,例如添加在线用户列表、私聊功能等。Django Channels 提供了强大的工具,让你可以构建出功能丰富的实时 Web 应用。