基于websocket+Redis的实时聊天系统

以下是一个简单的基于websocket和Redis的实时聊天系统的Python示例代码。

首先，安装必要的库：

pip install channels==3.0.4 django==3.1.5 redis==3.5.3

在你的Django项目的settings.py中添加以下配置：

# settings.py
 
INSTALLED_APPS = [
    # ...
    'channels',
    'chat',  # 假设你的应用名为chat
]
 
# Channel layer settings
CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {
            "hosts": [('127.0.0.1', 6379)],
        },
    },
}

在你的应用目录中创建routing.py文件，并添加以下内容：

# chat/routing.py
 
from channels.routing import ProtocolTypeRouter, URLRouter
from django.urls import path
from .consumers import chat_consumer
 
websocket_urlpatterns = [
    path('ws/chat/', chat_consumer),
]
 
channel_routing = ProtocolTypeRouter({
    "websocket": URLRouter(websocket_urlpatterns),
})

在你的应用目录中创建consumers.py文件，并添加以下内容：

# chat/consumers.py
 
import json
from channels.generic.websocket import AsyncWebsocketConsumer
from asgiref.sync import async_to_sync
 
class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        await self.channel_layer.group_add('chat', self.channel_name)
        await self.accept()
 
    async def disconnect(self, close_code):
        await self.channel_layer.group_discard('chat', 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(
            'chat',
            {
                'type': 'chat_message',
                'message': message,
            }
        )
 
    async def chat_message(self, event):
        message = event['message']
        await self.send(text_data=json.dumps({
            'message': message
        }))

在你的主urls.py文件中导入和添加websocket路由：

# your_project/urls.py
 
from django.urls import path, include
 
urlpatterns = [
    # ...
    path('ws/', include('chat.routing')),
]

这个简单的聊天系统使用了Django Channels库，通过websocket和Redis来实现实时通信。当WebSocket连接建立时，客户端被加入到名为'chat'的群组中。发送的消息被群组中的所有客户端接收。