使用 Django Channels 构建实时聊天应用（包含用户认证和消息持久化）

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.pyINSTALLED_APPS = [...'channels','chat',
]ASGI_APPLICATION = 'realtime_chat.routing.application'

创建一个新的 ASGI 路由文件 routing.py：

# routing.pyfrom channels.routing import ProtocolTypeRouterapplication = ProtocolTypeRouter({})

编写 Consumer

在 consumers.py 文件中编写一个简单的 Consumer：

# chat/consumers.pyfrom channels.generic.websocket import AsyncWebsocketConsumerclass ChatConsumer(AsyncWebsocketConsumer):async def connect(self):await self.accept()async def disconnect(self, close_code):passasync def receive(self, text_data):pass

编写路由

在 routing.py 文件中添加路由配置：

# routing.pyfrom channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from django.urls import path
from chat.consumers import ChatConsumerapplication = ProtocolTypeRouter({'websocket': AuthMiddlewareStack(URLRouter([path('chat/', ChatConsumer.as_asgi()),])),
})

创建 URL 路由

在 urls.py 文件中添加 URL 路由：

# chat/urls.pyfrom django.urls import path
from . import consumerswebsocket_urlpatterns = [path('chat/', consumers.ChatConsumer.as_asgi()),
]

在主项目的 urls.py 中包含 Chat 应用的路由：

# realtime_chat/urls.pyfrom django.urls import path, includeurlpatterns = [...path('chat/', include('chat.urls')),
]

运行应用

运行 Django 服务器：

python manage.py runserver

现在，你可以通过访问 http://localhost:8000/chat/ 来测试你的实时聊天应用了。

用户认证

首先，我们将使用 Django 自带的认证系统来处理用户认证。在 settings.py 中启用认证系统：

# settings.pyINSTALLED_APPS = [...'django.contrib.auth','django.contrib.contenttypes',...
]AUTHENTICATION_BACKENDS = ['django.contrib.auth.backends.ModelBackend',
]

然后，我们需要修改 ChatConsumer，以便处理用户认证：

# chat/consumers.pyfrom channels.generic.websocket import AsyncWebsocketConsumer
from django.contrib.auth.models import AnonymousUserclass 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):passasync def receive(self, text_data):pass

现在，只有经过身份验证的用户才能连接到聊天消费者。

消息持久化

我们希望用户在重新加载页面后能够看到之前的聊天消息。为了实现这一点，我们将使用 Django 的数据库来存储消息。

首先，我们需要创建一个模型来存储聊天消息：

# chat/models.pyfrom django.db import models
from django.contrib.auth.models import Userclass 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.pyfrom channels.generic.websocket import AsyncWebsocketConsumer
from django.contrib.auth.models import AnonymousUser
from .models import Message
import jsonclass 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):passasync 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.pyclass 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))@staticmethoddef 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.pyasync def connect(self):...await self.fetch_messages()

在前端，你可以使用 JavaScript 来接收并显示历史消息。

结论

通过添加用户认证和消息持久化功能，我们的实时聊天应用变得更加完善和实用。你可以根据需要进一步定制和扩展这个应用，例如添加在线用户列表、私聊功能等。Django Channels 提供了强大的工具，让你可以构建出功能丰富的实时 Web 应用。