简介这是一套基于Python全栈技术实现的轻量级多人实时聊天系统面向Web开发初学者与Django进阶学习者解决在线通信场景下的用户管理、状态同步与消息实时推送等核心问题。资源包共80个文件包含21个Python源码涵盖Django应用、路由、消费者及模型逻辑、7个HTML模板页登录、注册、聊天界面等、6个CSS与4个JS前端脚本含jQuery与Bootstrap支持以及字体、图标、截图和LICENSE等配套资源整体压缩包仅1.78MB结构清晰、开箱即用。已有1349人学习下载适合快速理解Django Channels异步通信机制、Redis在线状态管理、MySQL用户数据持久化及前后端协同开发流程。项目提供完整可运行示例含用户注册登录、密码修改、在线/离线用户列表、进出广播通知及无延时消息交互所有功能均通过真实页面截图与README说明验证便于对照调试与二次开发。1. 为什么一个“简单多人聊天系统”必须同时用上 Django、MySQL 和 Redis你可能试过用 Flask 写个带 WebSocket 的聊天页页面一刷消息就丢也可能用 Django Channels 做了实时推送但用户量刚过 50CPU 就飙到 90%更常见的是——消息能发、能收但离线用户上线后收不到历史记录或者两人同时发消息出现顺序错乱。这不是代码写得不够快而是架构层面对「状态」和「时序」的处理没分清边界。Django 负责用户管理、会话控制、HTTP 路由和后台管理MySQL 存用户资料、群组关系、已发送的每一条消息含时间戳、发送者 ID、接收目标而 Redis 不是来凑数的它承担在线状态标记user:123:online、未读计数缓存chat:456:unread:789、消息广播队列queue:group:456以及 Session 后端加速。三者分工明确MySQL 是唯一可信的消息持久化源Redis 是实时性保障的加速器Django 是粘合剂与业务逻辑中枢。这个组合不追求高并发百万级但能让 200 人以内团队内部聊天系统在普通云服务器2C4G上稳定运行且后续可平滑接入消息撤回、已读回执、搜索历史等需求——前提是从第一天起就把数据该存哪、该缓存什么、该由谁触发同步想清楚。2. 搭建基础环境Python 3.10、Django 4.2、MySQL 8.0 与 Redis 7 的最小可行配置2.1 环境版本选择依据与安装验证选择 Python 3.10 是因 Django 4.2 已放弃对 3.9 及以下的支持且 3.10 引入的结构化模式匹配match/case在后续扩展消息类型解析时有实际价值Django 4.2 是当前 LTS 版本长期安全更新支持至 2026 年 4 月MySQL 8.0 启用默认的caching_sha2_password认证插件需显式配置mysqlclient兼容Redis 7 支持原生的 Streams 数据结构比 List 更适合做消息队列天然支持消费者组、消息确认、历史回溯。安装命令需带版本锁定避免 CI/CD 环境中因自动升级导致兼容问题# 创建隔离环境推荐使用 venv非 conda python3.10 -m venv chat_env source chat_env/bin/activate # Linux/macOS # chat_env\Scripts\activate.bat # Windows # 安装指定版本依赖注意 mysqlclient 需先装系统依赖 pip install Django4.2.18 mysqlclient2.2.4 redis4.6.0 channels4.0.0 asgiref3.7.2提示mysqlclient2.2.4是目前与 MySQL 8.0.33 兼容最稳定的版本若pip install报mysql_config not found错误请先执行sudo apt-get install default-libmysqlclient-dev build-essentialUbuntu/Debian或brew install mysql-clientmacOS。2.2 MySQL 初始化建库、建表、设权限的三步命令聊天系统不需要复杂分库分表但必须为 Django ORM 和手写 SQL 预留扩展空间。创建专用数据库chat_db字符集强制设为utf8mb4支持 emoji并分配最小权限账号chat_user-- 登录 MySQL如 root mysql -u root -p -- 创建数据库显式指定字符集和排序规则 CREATE DATABASE chat_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; -- 创建应用专用账号限制仅本地连接密码强度需符合 MySQL 8.0 默认策略 CREATE USER chat_userlocalhost IDENTIFIED BY StrongPssw0rd2024!; -- 授予仅必要权限不给 DROP、GRANT OPTION GRANT SELECT, INSERT, UPDATE, DELETE ON chat_db.* TO chat_userlocalhost; -- 刷新权限 FLUSH PRIVILEGES;验证是否生效mysql -u chat_user -p -D chat_db -e SELECT VERSION(); # 应返回 MySQL 版本号且无 ERROR 10452.3 Redis 配置要点禁用 AOF、启用内存淘汰、绑定本地地址生产环境不建议直接用默认配置跑 Redis。聊天系统对消息丢失容忍度低Redis 仅作缓存/队列非主存储但对响应延迟敏感。关键修改项如下编辑/etc/redis/redis.conf或 Docker 启动参数配置项推荐值说明bind 127.0.0.1 ::1必须设置禁止公网暴露仅允许本地 Django 进程访问port 6379保持默认若端口被占改后需同步更新 Djangosettings.pymaxmemory 512mb根据服务器内存调整防止 Redis 占满内存导致 OOM Killer 杀进程maxmemory-policy allkeys-lru必须设置当内存满时优先淘汰最久未用的 key保障在线状态等热数据不被淘汰appendonly no关键聊天消息由 MySQL 持久化Redis 不开启 AOF降低写延迟save 清空所有 save 规则禁用 RDB 快照避免 fork 阻塞主线程启动并验证# Ubuntu 系统服务方式 sudo systemctl restart redis-server sudo systemctl status redis-server # 确认 active (running) # 测试连通性 redis-cli -h 127.0.0.1 -p 6379 ping # 应返回 PONG3. Django 项目结构设计从models.py到consumers.py的四层数据流3.1 数据模型定义User、ChatRoom、Message 的关联与索引优化Django 自带auth.User模型足够支撑用户认证但需扩展在线状态字段ChatRoom表支持私聊typeprivate与群聊typegroupMessage表必须包含created_at用于排序、is_read初始为 False、sender和receiver私聊时存对方 ID群聊时receiver为空用room字段关联。关键点在于索引——MySQL 查询消息列表时WHERE room_id ? ORDER BY created_at DESC LIMIT 20是高频操作必须为(room_id, created_at)建联合索引# chat/models.py from django.db import models from django.contrib.auth.models import User class ChatRoom(models.Model): name models.CharField(max_length100) type models.CharField(max_length10, choices[(private, 私聊), (group, 群聊)]) created_at models.DateTimeField(auto_now_addTrue) def __str__(self): return f{self.name} ({self.type}) class Message(models.Model): sender models.ForeignKey(User, on_deletemodels.CASCADE, related_namesent_messages) receiver models.ForeignKey(User, on_deletemodels.SET_NULL, nullTrue, blankTrue, related_namereceived_messages) room models.ForeignKey(ChatRoom, on_deletemodels.CASCADE, related_namemessages) content models.TextField() is_read models.BooleanField(defaultFalse) created_at models.DateTimeField(auto_now_addTrue) class Meta: # 关键为查询性能加联合索引 indexes [ models.Index(fields[room, -created_at]), # room_id 降序 created_at models.Index(fields[sender, -created_at]), ] ordering [-created_at] # 默认按时间倒序 def __str__(self): return f{self.sender.username} → {self.content[:20]}执行迁移python manage.py makemigrations python manage.py migrate注意makemigrations生成的 SQL 中CREATE INDEX ... ON chat_message (room_id, created_at DESC)在 MySQL 8.0 才支持DESC若报错请手动修改迁移文件去掉DESCMySQL 会自动按索引顺序优化查询。3.2 Django Channels 配置ASGI、Routing 与 Consumer 的三层路由Django 4.2 默认启用 ASGI需替换wsgi.py为asgi.py并配置routing.py。WebSocket 连接路径/ws/chat/room_id/由ChatConsumer处理该 Consumer 继承AsyncWebsocketConsumer实现连接、接收、断开三方法。重点在于每个用户连接必须加入对应房间的 Channel Group且 Group 名需全局唯一用fchat_{room_id}# chat/asgi.py import os from django.core.asgi import get_asgi_application from channels.routing import ProtocolTypeRouter, URLRouter from channels.auth import AuthMiddlewareStack from django.urls import path from chat import consumers os.environ.setdefault(DJANGO_SETTINGS_MODULE, chat_system.settings) application ProtocolTypeRouter({ http: get_asgi_application(), websocket: AuthMiddlewareStack( URLRouter([ path(ws/chat/int:room_id/, consumers.ChatConsumer.as_asgi()), ]) ), })# chat/consumers.py import json from channels.generic.websocket import AsyncWebsocketConsumer from channels.db import database_sync_to_async from django.contrib.auth.models import User from .models import ChatRoom, Message class ChatConsumer(AsyncWebsocketConsumer): async def connect(self): self.room_id self.scope[url_route][kwargs][room_id] self.room_group_name fchat_{self.room_id} # 加入房间 Group await self.channel_layer.group_add( self.room_group_name, self.channel_name ) await self.accept() async def disconnect(self, close_code): # 断开时离开 Group 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] sender_id self.scope[user].id # 异步保存消息到 MySQL msg_obj await self.save_message(sender_id, message) # 广播给房间内所有客户端包括自己 await self.channel_layer.group_send( self.room_group_name, { type: chat_message, message: message, sender: self.scope[user].username, timestamp: msg_obj.created_at.isoformat(), message_id: msg_obj.id, } ) async def chat_message(self, event): # 向 WebSocket 发送消息 await self.send(text_datajson.dumps({ message: event[message], sender: event[sender], timestamp: event[timestamp], message_id: event[message_id], })) database_sync_to_async def save_message(self, sender_id, content): sender User.objects.get(idsender_id) room ChatRoom.objects.get(idself.room_id) return Message.objects.create( sendersender, roomroom, contentcontent )3.3 Redis 集成用redis-py实现在线状态与未读计数的原子操作Django Channels 的channel_layer默认用 Redis 做后端但业务逻辑中的在线状态user:123:online和未读数chat:456:unread:789需直接调用redis-py。关键要求状态变更必须原子避免竞态。例如用户上线时需同时设置在线状态EX 300 秒并发布上线事件用户发消息后需对每个未读成员的计数INCR# chat/utils.py import redis from django.conf import settings # 初始化 Redis 连接池复用连接避免频繁创建 redis_client redis.Redis( hostsettings.REDIS_HOST, portsettings.REDIS_PORT, dbsettings.REDIS_DB, decode_responsesTrue, # 自动解码 bytes 为 str socket_connect_timeout2, socket_timeout2 ) def set_user_online(user_id: int) - bool: 设置用户在线状态5分钟过期 key fuser:{user_id}:online try: # SET key value EX seconds NX仅当 key 不存在时设置防止重复上线覆盖 result redis_client.set(key, 1, ex300, nxTrue) if result: # 发布上线事件供其他服务监听如通知好友 redis_client.publish(user_status, json.dumps({user_id: user_id, status: online})) return bool(result) except redis.ConnectionError: return False def incr_unread_count(room_id: int, target_user_id: int) - int: 为指定用户增加未读消息数返回新值 key fchat:{room_id}:unread:{target_user_id} try: # INCR 原子自增不存在则初始化为 0 再加 1 return redis_client.incr(key) except redis.ConnectionError: return 0 def get_unread_count(room_id: int, user_id: int) - int: 获取未读数不存在返回 0 key fchat:{room_id}:unread:{user_id} val redis_client.get(key) return int(val) if val else 0 def mark_message_as_read(message_id: int, user_id: int): 标记单条消息为已读并减少未读计数 try: # 先查消息所属房间需在 Message 模型中加 room_id 字段冗余或 JOIN 查询 msg Message.objects.select_related(room).get(idmessage_id) # 减少未读计数DECR key fchat:{msg.room_id}:unread:{user_id} redis_client.decr(key) # 更新数据库标记 Message.objects.filter(idmessage_id).update(is_readTrue) except (Message.DoesNotExist, redis.ConnectionError): pass在settings.py中添加 Redis 配置# chat_system/settings.py REDIS_HOST 127.0.0.1 REDIS_PORT 6379 REDIS_DB 1 # 与 channel_layer 分开避免冲突4. 前端实时交互实现Vue 3 WebSocket 的消息收发与状态同步4.1 Vue 3 Composition API 构建聊天组件useWebSocket 与 useOnlineStatus前端不依赖 jQuery 或老旧框架用 Vue 3 的ref、onMounted和onUnmounted管理 WebSocket 生命周期。核心逻辑连接成功后立即拉取历史消息HTTP GET再监听 WebSocket 事件用户输入回车即发送收到新消息时自动滚动到底部。在线状态通过 Redis Pub/Sub 监听user_status频道实现!-- src/components/ChatRoom.vue -- template div classchat-container div classchat-header h2{{ roomName }}/h2 div classonline-status span :class{ online: isOnline }●/span {{ isOnline ? 在线 : 离线 }} /div /div div refmessagesContainer classmessages-list div v-formsg in messages :keymsg.id classmessage strong{{ msg.sender }}:/strong span{{ msg.message }}/span small{{ formatTime(msg.timestamp) }}/small /div /div div classinput-area input v-modelnewMessage keyup.entersendMessage placeholder输入消息... classmessage-input / button clicksendMessage classsend-btn发送/button /div /div /template script setup import { ref, onMounted, onUnmounted, watch } from vue import { formatDistanceToNow } from date-fns const props defineProps({ roomId: { type: Number, required: true }, roomName: { type: String, required: true } }) const messages ref([]) const newMessage ref() const messagesContainer ref(null) const isOnline ref(true) // 初始设为在线后续由 Redis Pub/Sub 更新 // WebSocket 实例 let socket null // 连接 WebSocket const connectWebSocket () { const wsUrl ws://${window.location.host}/ws/chat/${props.roomId}/ socket new WebSocket(wsUrl) socket.onopen () { console.log(WebSocket connected) } socket.onmessage (event) { const data JSON.parse(event.data) messages.value.push(data) // 滚动到底部 messagesContainer.value?.scrollIntoView({ behavior: smooth, block: end }) } socket.onerror (error) { console.error(WebSocket error:, error) } socket.onclose () { console.log(WebSocket disconnected) } } // 发送消息 const sendMessage () { if (!newMessage.value.trim() || !socket || socket.readyState ! WebSocket.OPEN) return socket.send(JSON.stringify({ message: newMessage.value.trim() })) newMessage.value } // 获取历史消息HTTP const fetchHistory async () { try { const res await fetch(/api/messages/?room_id${props.roomId}limit20) const data await res.json() messages.value data } catch (err) { console.error(Fetch history failed:, err) } } // 格式化时间 const formatTime (isoString) { return formatDistanceToNow(new Date(isoString), { addSuffix: true }) } // 监听 Redis Pub/Sub 在线状态需后端提供 /api/user/status 接口轮询或 SSE const checkOnlineStatus async () { try { const res await fetch(/api/user/status/?user_id${props.roomId}) const data await res.json() isOnline.value data.status online } catch (err) { isOnline.value false } } onMounted(() { fetchHistory() connectWebSocket() // 每30秒检查一次在线状态简化版生产可用 SSE 替代 const statusInterval setInterval(checkOnlineStatus, 30000) onUnmounted(() clearInterval(statusInterval)) }) // 组件卸载时关闭连接 onUnmounted(() { if (socket socket.readyState WebSocket.OPEN) { socket.close() } }) /script4.2 Django REST API 提供消息历史与状态查询接口WebSocket 负责实时推送HTTP API 负责补全历史与状态。/api/messages/返回分页消息按created_at倒序/api/user/status/查询 Redis 中的在线标记# chat/views.py from django.http import JsonResponse from django.views.decorators.http import require_http_methods from django.views.decorators.csrf import csrf_exempt from django.contrib.auth.decorators import login_required from django.core.paginator import Paginator from django.utils import timezone from .models import Message, ChatRoom from .utils import get_unread_count, set_user_online login_required require_http_methods([GET]) def get_messages(request): room_id request.GET.get(room_id) limit int(request.GET.get(limit, 20)) if not room_id: return JsonResponse({error: room_id required}, status400) messages Message.objects.filter( room_idroom_id ).select_related(sender).values( id, sender__username, content, created_at, is_read ).order_by(-created_at)[:limit] # 转换为标准 JSON 格式 msg_list [] for m in messages: msg_list.append({ id: m[id], sender: m[sender__username], message: m[content], timestamp: m[created_at].isoformat(), is_read: m[is_read] }) return JsonResponse(msg_list, safeFalse) login_required require_http_methods([GET]) def user_status(request): user_id request.GET.get(user_id) if not user_id: return JsonResponse({error: user_id required}, status400) # 查询 Redis 中的在线状态 key fuser:{user_id}:online is_online bool(redis_client.get(key)) return JsonResponse({ user_id: user_id, status: online if is_online else offline, last_seen: timezone.now().isoformat() if is_online else None })URL 配置# chat/urls.py from django.urls import path from . import views urlpatterns [ path(api/messages/, views.get_messages, nameget_messages), path(api/user/status/, views.user_status, nameuser_status), ]5. 生产部署与关键参数调优Nginx 反向代理、Supervisor 进程管理与 Redis 内存监控5.1 Nginx 配置 WebSocket 透传与静态资源分离Django 开发服务器不支持 WebSocket 长连接必须用 Nginx 做反向代理。关键点Upgrade和Connection头必须透传否则 WebSocket 握手失败静态文件CSS/JS由 Nginx 直接服务减轻 Django 压力# /etc/nginx/sites-available/chat-system upstream django_app { server 127.0.0.1:8000; # Django runserver 或 Gunicorn } upstream daphne_app { server 127.0.0.1:8001; # Daphne ASGI 服务器 } server { listen 80; server_name chat.example.com; # 静态文件由 Nginx 直接服务 location /static/ { alias /home/ubuntu/chat_system/staticfiles/; expires 1y; add_header Cache-Control public, immutable; } # WebSocket 路径透传 location /ws/ { proxy_pass http://daphne_app; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_redirect off; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; } # 其他请求走 Django location / { proxy_pass http://django_app; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; } }启用配置sudo ln -sf /etc/nginx/sites-available/chat-system /etc/nginx/sites-enabled/ sudo nginx -t sudo systemctl reload nginx5.2 Supervisor 管理 Daphne 与 Redis 进程确保崩溃自动重启Daphne 是 Django Channels 官方推荐的 ASGI 服务器需与 Redis 一起由 Supervisor 管理避免手动启停遗漏# /etc/supervisor/conf.d/chat-system.conf [program:daphne] command/home/ubuntu/chat_env/bin/daphne -b 127.0.0.1 -p 8001 chat_system.asgi:application directory/home/ubuntu/chat_system userubuntu autostarttrue autorestarttrue redirect_stderrtrue stdout_logfile/var/log/chat-system/daphne.log [program:redis] command/usr/bin/redis-server /etc/redis/redis.conf userredis autostarttrue autorestarttrue redirect_stderrtrue stdout_logfile/var/log/chat-system/redis.log [group:chat] programsdaphne,redis重载 Supervisorsudo supervisorctl reread sudo supervisorctl update sudo supervisorctl start chat:*5.3 Redis 内存使用监控识别大 Key 与内存泄漏的三个命令Redis 内存异常增长是聊天系统最常见故障。用以下命令快速定位查看内存总用量与峰值redis-cli info memory | grep -E (used_memory_human|used_memory_peak_human|mem_fragmentation_ratio) # 输出示例used_memory_human:124.50M, used_memory_peak_human:130.20M, mem_fragmentation_ratio:1.02扫描大 Key10KBredis-cli --bigkeys # 输出类似Biggest string found user:123:profile with 12543 bytes查看某类 Key 的数量与平均大小# 统计所有 chat:*:unread:* 类型的 Key 数量 redis-cli eval return #redis.call(keys, chat:*:unread:*) 0 # 查看前 10 个 unread Key 的大小 redis-cli --scan --pattern chat:*:unread:* | head -10 | xargs -I {} redis-cli memory usage {}提示若发现user:*:online类 Key 持续增长检查set_user_online是否漏掉nxTrue参数导致每次上线都新建 key 而非覆盖若chat:*:unread:*数量爆炸需确认消息已读回调是否正确调用mark_message_as_read。部署完成后用ab或wrk做基础压测# 模拟 50 并发、持续 30 秒的 WebSocket 连接需用专门工具如 wstest # 或测试 HTTP API wrk -t4 -c50 -d30s http://localhost/api/messages/?room_id1正常响应时间应稳定在 20ms 内RedisINFO memory中mem_fragmentation_ratio在 1.0~1.5 之间为健康。本文还有配套的精品资源点击获取 Django+MySQL+Redis构建稳定聊天系统架构
简介这是一套基于Python全栈技术实现的轻量级多人实时聊天系统面向Web开发初学者与Django进阶学习者解决在线通信场景下的用户管理、状态同步与消息实时推送等核心问题。资源包共80个文件包含21个Python源码涵盖Django应用、路由、消费者及模型逻辑、7个HTML模板页登录、注册、聊天界面等、6个CSS与4个JS前端脚本含jQuery与Bootstrap支持以及字体、图标、截图和LICENSE等配套资源整体压缩包仅1.78MB结构清晰、开箱即用。已有1349人学习下载适合快速理解Django Channels异步通信机制、Redis在线状态管理、MySQL用户数据持久化及前后端协同开发流程。项目提供完整可运行示例含用户注册登录、密码修改、在线/离线用户列表、进出广播通知及无延时消息交互所有功能均通过真实页面截图与README说明验证便于对照调试与二次开发。1. 为什么一个“简单多人聊天系统”必须同时用上 Django、MySQL 和 Redis你可能试过用 Flask 写个带 WebSocket 的聊天页页面一刷消息就丢也可能用 Django Channels 做了实时推送但用户量刚过 50CPU 就飙到 90%更常见的是——消息能发、能收但离线用户上线后收不到历史记录或者两人同时发消息出现顺序错乱。这不是代码写得不够快而是架构层面对「状态」和「时序」的处理没分清边界。Django 负责用户管理、会话控制、HTTP 路由和后台管理MySQL 存用户资料、群组关系、已发送的每一条消息含时间戳、发送者 ID、接收目标而 Redis 不是来凑数的它承担在线状态标记user:123:online、未读计数缓存chat:456:unread:789、消息广播队列queue:group:456以及 Session 后端加速。三者分工明确MySQL 是唯一可信的消息持久化源Redis 是实时性保障的加速器Django 是粘合剂与业务逻辑中枢。这个组合不追求高并发百万级但能让 200 人以内团队内部聊天系统在普通云服务器2C4G上稳定运行且后续可平滑接入消息撤回、已读回执、搜索历史等需求——前提是从第一天起就把数据该存哪、该缓存什么、该由谁触发同步想清楚。2. 搭建基础环境Python 3.10、Django 4.2、MySQL 8.0 与 Redis 7 的最小可行配置2.1 环境版本选择依据与安装验证选择 Python 3.10 是因 Django 4.2 已放弃对 3.9 及以下的支持且 3.10 引入的结构化模式匹配match/case在后续扩展消息类型解析时有实际价值Django 4.2 是当前 LTS 版本长期安全更新支持至 2026 年 4 月MySQL 8.0 启用默认的caching_sha2_password认证插件需显式配置mysqlclient兼容Redis 7 支持原生的 Streams 数据结构比 List 更适合做消息队列天然支持消费者组、消息确认、历史回溯。安装命令需带版本锁定避免 CI/CD 环境中因自动升级导致兼容问题# 创建隔离环境推荐使用 venv非 conda python3.10 -m venv chat_env source chat_env/bin/activate # Linux/macOS # chat_env\Scripts\activate.bat # Windows # 安装指定版本依赖注意 mysqlclient 需先装系统依赖 pip install Django4.2.18 mysqlclient2.2.4 redis4.6.0 channels4.0.0 asgiref3.7.2提示mysqlclient2.2.4是目前与 MySQL 8.0.33 兼容最稳定的版本若pip install报mysql_config not found错误请先执行sudo apt-get install default-libmysqlclient-dev build-essentialUbuntu/Debian或brew install mysql-clientmacOS。2.2 MySQL 初始化建库、建表、设权限的三步命令聊天系统不需要复杂分库分表但必须为 Django ORM 和手写 SQL 预留扩展空间。创建专用数据库chat_db字符集强制设为utf8mb4支持 emoji并分配最小权限账号chat_user-- 登录 MySQL如 root mysql -u root -p -- 创建数据库显式指定字符集和排序规则 CREATE DATABASE chat_db CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; -- 创建应用专用账号限制仅本地连接密码强度需符合 MySQL 8.0 默认策略 CREATE USER chat_userlocalhost IDENTIFIED BY StrongPssw0rd2024!; -- 授予仅必要权限不给 DROP、GRANT OPTION GRANT SELECT, INSERT, UPDATE, DELETE ON chat_db.* TO chat_userlocalhost; -- 刷新权限 FLUSH PRIVILEGES;验证是否生效mysql -u chat_user -p -D chat_db -e SELECT VERSION(); # 应返回 MySQL 版本号且无 ERROR 10452.3 Redis 配置要点禁用 AOF、启用内存淘汰、绑定本地地址生产环境不建议直接用默认配置跑 Redis。聊天系统对消息丢失容忍度低Redis 仅作缓存/队列非主存储但对响应延迟敏感。关键修改项如下编辑/etc/redis/redis.conf或 Docker 启动参数配置项推荐值说明bind 127.0.0.1 ::1必须设置禁止公网暴露仅允许本地 Django 进程访问port 6379保持默认若端口被占改后需同步更新 Djangosettings.pymaxmemory 512mb根据服务器内存调整防止 Redis 占满内存导致 OOM Killer 杀进程maxmemory-policy allkeys-lru必须设置当内存满时优先淘汰最久未用的 key保障在线状态等热数据不被淘汰appendonly no关键聊天消息由 MySQL 持久化Redis 不开启 AOF降低写延迟save 清空所有 save 规则禁用 RDB 快照避免 fork 阻塞主线程启动并验证# Ubuntu 系统服务方式 sudo systemctl restart redis-server sudo systemctl status redis-server # 确认 active (running) # 测试连通性 redis-cli -h 127.0.0.1 -p 6379 ping # 应返回 PONG3. Django 项目结构设计从models.py到consumers.py的四层数据流3.1 数据模型定义User、ChatRoom、Message 的关联与索引优化Django 自带auth.User模型足够支撑用户认证但需扩展在线状态字段ChatRoom表支持私聊typeprivate与群聊typegroupMessage表必须包含created_at用于排序、is_read初始为 False、sender和receiver私聊时存对方 ID群聊时receiver为空用room字段关联。关键点在于索引——MySQL 查询消息列表时WHERE room_id ? ORDER BY created_at DESC LIMIT 20是高频操作必须为(room_id, created_at)建联合索引# chat/models.py from django.db import models from django.contrib.auth.models import User class ChatRoom(models.Model): name models.CharField(max_length100) type models.CharField(max_length10, choices[(private, 私聊), (group, 群聊)]) created_at models.DateTimeField(auto_now_addTrue) def __str__(self): return f{self.name} ({self.type}) class Message(models.Model): sender models.ForeignKey(User, on_deletemodels.CASCADE, related_namesent_messages) receiver models.ForeignKey(User, on_deletemodels.SET_NULL, nullTrue, blankTrue, related_namereceived_messages) room models.ForeignKey(ChatRoom, on_deletemodels.CASCADE, related_namemessages) content models.TextField() is_read models.BooleanField(defaultFalse) created_at models.DateTimeField(auto_now_addTrue) class Meta: # 关键为查询性能加联合索引 indexes [ models.Index(fields[room, -created_at]), # room_id 降序 created_at models.Index(fields[sender, -created_at]), ] ordering [-created_at] # 默认按时间倒序 def __str__(self): return f{self.sender.username} → {self.content[:20]}执行迁移python manage.py makemigrations python manage.py migrate注意makemigrations生成的 SQL 中CREATE INDEX ... ON chat_message (room_id, created_at DESC)在 MySQL 8.0 才支持DESC若报错请手动修改迁移文件去掉DESCMySQL 会自动按索引顺序优化查询。3.2 Django Channels 配置ASGI、Routing 与 Consumer 的三层路由Django 4.2 默认启用 ASGI需替换wsgi.py为asgi.py并配置routing.py。WebSocket 连接路径/ws/chat/room_id/由ChatConsumer处理该 Consumer 继承AsyncWebsocketConsumer实现连接、接收、断开三方法。重点在于每个用户连接必须加入对应房间的 Channel Group且 Group 名需全局唯一用fchat_{room_id}# chat/asgi.py import os from django.core.asgi import get_asgi_application from channels.routing import ProtocolTypeRouter, URLRouter from channels.auth import AuthMiddlewareStack from django.urls import path from chat import consumers os.environ.setdefault(DJANGO_SETTINGS_MODULE, chat_system.settings) application ProtocolTypeRouter({ http: get_asgi_application(), websocket: AuthMiddlewareStack( URLRouter([ path(ws/chat/int:room_id/, consumers.ChatConsumer.as_asgi()), ]) ), })# chat/consumers.py import json from channels.generic.websocket import AsyncWebsocketConsumer from channels.db import database_sync_to_async from django.contrib.auth.models import User from .models import ChatRoom, Message class ChatConsumer(AsyncWebsocketConsumer): async def connect(self): self.room_id self.scope[url_route][kwargs][room_id] self.room_group_name fchat_{self.room_id} # 加入房间 Group await self.channel_layer.group_add( self.room_group_name, self.channel_name ) await self.accept() async def disconnect(self, close_code): # 断开时离开 Group 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] sender_id self.scope[user].id # 异步保存消息到 MySQL msg_obj await self.save_message(sender_id, message) # 广播给房间内所有客户端包括自己 await self.channel_layer.group_send( self.room_group_name, { type: chat_message, message: message, sender: self.scope[user].username, timestamp: msg_obj.created_at.isoformat(), message_id: msg_obj.id, } ) async def chat_message(self, event): # 向 WebSocket 发送消息 await self.send(text_datajson.dumps({ message: event[message], sender: event[sender], timestamp: event[timestamp], message_id: event[message_id], })) database_sync_to_async def save_message(self, sender_id, content): sender User.objects.get(idsender_id) room ChatRoom.objects.get(idself.room_id) return Message.objects.create( sendersender, roomroom, contentcontent )3.3 Redis 集成用redis-py实现在线状态与未读计数的原子操作Django Channels 的channel_layer默认用 Redis 做后端但业务逻辑中的在线状态user:123:online和未读数chat:456:unread:789需直接调用redis-py。关键要求状态变更必须原子避免竞态。例如用户上线时需同时设置在线状态EX 300 秒并发布上线事件用户发消息后需对每个未读成员的计数INCR# chat/utils.py import redis from django.conf import settings # 初始化 Redis 连接池复用连接避免频繁创建 redis_client redis.Redis( hostsettings.REDIS_HOST, portsettings.REDIS_PORT, dbsettings.REDIS_DB, decode_responsesTrue, # 自动解码 bytes 为 str socket_connect_timeout2, socket_timeout2 ) def set_user_online(user_id: int) - bool: 设置用户在线状态5分钟过期 key fuser:{user_id}:online try: # SET key value EX seconds NX仅当 key 不存在时设置防止重复上线覆盖 result redis_client.set(key, 1, ex300, nxTrue) if result: # 发布上线事件供其他服务监听如通知好友 redis_client.publish(user_status, json.dumps({user_id: user_id, status: online})) return bool(result) except redis.ConnectionError: return False def incr_unread_count(room_id: int, target_user_id: int) - int: 为指定用户增加未读消息数返回新值 key fchat:{room_id}:unread:{target_user_id} try: # INCR 原子自增不存在则初始化为 0 再加 1 return redis_client.incr(key) except redis.ConnectionError: return 0 def get_unread_count(room_id: int, user_id: int) - int: 获取未读数不存在返回 0 key fchat:{room_id}:unread:{user_id} val redis_client.get(key) return int(val) if val else 0 def mark_message_as_read(message_id: int, user_id: int): 标记单条消息为已读并减少未读计数 try: # 先查消息所属房间需在 Message 模型中加 room_id 字段冗余或 JOIN 查询 msg Message.objects.select_related(room).get(idmessage_id) # 减少未读计数DECR key fchat:{msg.room_id}:unread:{user_id} redis_client.decr(key) # 更新数据库标记 Message.objects.filter(idmessage_id).update(is_readTrue) except (Message.DoesNotExist, redis.ConnectionError): pass在settings.py中添加 Redis 配置# chat_system/settings.py REDIS_HOST 127.0.0.1 REDIS_PORT 6379 REDIS_DB 1 # 与 channel_layer 分开避免冲突4. 前端实时交互实现Vue 3 WebSocket 的消息收发与状态同步4.1 Vue 3 Composition API 构建聊天组件useWebSocket 与 useOnlineStatus前端不依赖 jQuery 或老旧框架用 Vue 3 的ref、onMounted和onUnmounted管理 WebSocket 生命周期。核心逻辑连接成功后立即拉取历史消息HTTP GET再监听 WebSocket 事件用户输入回车即发送收到新消息时自动滚动到底部。在线状态通过 Redis Pub/Sub 监听user_status频道实现!-- src/components/ChatRoom.vue -- template div classchat-container div classchat-header h2{{ roomName }}/h2 div classonline-status span :class{ online: isOnline }●/span {{ isOnline ? 在线 : 离线 }} /div /div div refmessagesContainer classmessages-list div v-formsg in messages :keymsg.id classmessage strong{{ msg.sender }}:/strong span{{ msg.message }}/span small{{ formatTime(msg.timestamp) }}/small /div /div div classinput-area input v-modelnewMessage keyup.entersendMessage placeholder输入消息... classmessage-input / button clicksendMessage classsend-btn发送/button /div /div /template script setup import { ref, onMounted, onUnmounted, watch } from vue import { formatDistanceToNow } from date-fns const props defineProps({ roomId: { type: Number, required: true }, roomName: { type: String, required: true } }) const messages ref([]) const newMessage ref() const messagesContainer ref(null) const isOnline ref(true) // 初始设为在线后续由 Redis Pub/Sub 更新 // WebSocket 实例 let socket null // 连接 WebSocket const connectWebSocket () { const wsUrl ws://${window.location.host}/ws/chat/${props.roomId}/ socket new WebSocket(wsUrl) socket.onopen () { console.log(WebSocket connected) } socket.onmessage (event) { const data JSON.parse(event.data) messages.value.push(data) // 滚动到底部 messagesContainer.value?.scrollIntoView({ behavior: smooth, block: end }) } socket.onerror (error) { console.error(WebSocket error:, error) } socket.onclose () { console.log(WebSocket disconnected) } } // 发送消息 const sendMessage () { if (!newMessage.value.trim() || !socket || socket.readyState ! WebSocket.OPEN) return socket.send(JSON.stringify({ message: newMessage.value.trim() })) newMessage.value } // 获取历史消息HTTP const fetchHistory async () { try { const res await fetch(/api/messages/?room_id${props.roomId}limit20) const data await res.json() messages.value data } catch (err) { console.error(Fetch history failed:, err) } } // 格式化时间 const formatTime (isoString) { return formatDistanceToNow(new Date(isoString), { addSuffix: true }) } // 监听 Redis Pub/Sub 在线状态需后端提供 /api/user/status 接口轮询或 SSE const checkOnlineStatus async () { try { const res await fetch(/api/user/status/?user_id${props.roomId}) const data await res.json() isOnline.value data.status online } catch (err) { isOnline.value false } } onMounted(() { fetchHistory() connectWebSocket() // 每30秒检查一次在线状态简化版生产可用 SSE 替代 const statusInterval setInterval(checkOnlineStatus, 30000) onUnmounted(() clearInterval(statusInterval)) }) // 组件卸载时关闭连接 onUnmounted(() { if (socket socket.readyState WebSocket.OPEN) { socket.close() } }) /script4.2 Django REST API 提供消息历史与状态查询接口WebSocket 负责实时推送HTTP API 负责补全历史与状态。/api/messages/返回分页消息按created_at倒序/api/user/status/查询 Redis 中的在线标记# chat/views.py from django.http import JsonResponse from django.views.decorators.http import require_http_methods from django.views.decorators.csrf import csrf_exempt from django.contrib.auth.decorators import login_required from django.core.paginator import Paginator from django.utils import timezone from .models import Message, ChatRoom from .utils import get_unread_count, set_user_online login_required require_http_methods([GET]) def get_messages(request): room_id request.GET.get(room_id) limit int(request.GET.get(limit, 20)) if not room_id: return JsonResponse({error: room_id required}, status400) messages Message.objects.filter( room_idroom_id ).select_related(sender).values( id, sender__username, content, created_at, is_read ).order_by(-created_at)[:limit] # 转换为标准 JSON 格式 msg_list [] for m in messages: msg_list.append({ id: m[id], sender: m[sender__username], message: m[content], timestamp: m[created_at].isoformat(), is_read: m[is_read] }) return JsonResponse(msg_list, safeFalse) login_required require_http_methods([GET]) def user_status(request): user_id request.GET.get(user_id) if not user_id: return JsonResponse({error: user_id required}, status400) # 查询 Redis 中的在线状态 key fuser:{user_id}:online is_online bool(redis_client.get(key)) return JsonResponse({ user_id: user_id, status: online if is_online else offline, last_seen: timezone.now().isoformat() if is_online else None })URL 配置# chat/urls.py from django.urls import path from . import views urlpatterns [ path(api/messages/, views.get_messages, nameget_messages), path(api/user/status/, views.user_status, nameuser_status), ]5. 生产部署与关键参数调优Nginx 反向代理、Supervisor 进程管理与 Redis 内存监控5.1 Nginx 配置 WebSocket 透传与静态资源分离Django 开发服务器不支持 WebSocket 长连接必须用 Nginx 做反向代理。关键点Upgrade和Connection头必须透传否则 WebSocket 握手失败静态文件CSS/JS由 Nginx 直接服务减轻 Django 压力# /etc/nginx/sites-available/chat-system upstream django_app { server 127.0.0.1:8000; # Django runserver 或 Gunicorn } upstream daphne_app { server 127.0.0.1:8001; # Daphne ASGI 服务器 } server { listen 80; server_name chat.example.com; # 静态文件由 Nginx 直接服务 location /static/ { alias /home/ubuntu/chat_system/staticfiles/; expires 1y; add_header Cache-Control public, immutable; } # WebSocket 路径透传 location /ws/ { proxy_pass http://daphne_app; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; proxy_redirect off; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; } # 其他请求走 Django location / { proxy_pass http://django_app; proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; proxy_set_header X-Forwarded-Proto $scheme; } }启用配置sudo ln -sf /etc/nginx/sites-available/chat-system /etc/nginx/sites-enabled/ sudo nginx -t sudo systemctl reload nginx5.2 Supervisor 管理 Daphne 与 Redis 进程确保崩溃自动重启Daphne 是 Django Channels 官方推荐的 ASGI 服务器需与 Redis 一起由 Supervisor 管理避免手动启停遗漏# /etc/supervisor/conf.d/chat-system.conf [program:daphne] command/home/ubuntu/chat_env/bin/daphne -b 127.0.0.1 -p 8001 chat_system.asgi:application directory/home/ubuntu/chat_system userubuntu autostarttrue autorestarttrue redirect_stderrtrue stdout_logfile/var/log/chat-system/daphne.log [program:redis] command/usr/bin/redis-server /etc/redis/redis.conf userredis autostarttrue autorestarttrue redirect_stderrtrue stdout_logfile/var/log/chat-system/redis.log [group:chat] programsdaphne,redis重载 Supervisorsudo supervisorctl reread sudo supervisorctl update sudo supervisorctl start chat:*5.3 Redis 内存使用监控识别大 Key 与内存泄漏的三个命令Redis 内存异常增长是聊天系统最常见故障。用以下命令快速定位查看内存总用量与峰值redis-cli info memory | grep -E (used_memory_human|used_memory_peak_human|mem_fragmentation_ratio) # 输出示例used_memory_human:124.50M, used_memory_peak_human:130.20M, mem_fragmentation_ratio:1.02扫描大 Key10KBredis-cli --bigkeys # 输出类似Biggest string found user:123:profile with 12543 bytes查看某类 Key 的数量与平均大小# 统计所有 chat:*:unread:* 类型的 Key 数量 redis-cli eval return #redis.call(keys, chat:*:unread:*) 0 # 查看前 10 个 unread Key 的大小 redis-cli --scan --pattern chat:*:unread:* | head -10 | xargs -I {} redis-cli memory usage {}提示若发现user:*:online类 Key 持续增长检查set_user_online是否漏掉nxTrue参数导致每次上线都新建 key 而非覆盖若chat:*:unread:*数量爆炸需确认消息已读回调是否正确调用mark_message_as_read。部署完成后用ab或wrk做基础压测# 模拟 50 并发、持续 30 秒的 WebSocket 连接需用专门工具如 wstest # 或测试 HTTP API wrk -t4 -c50 -d30s http://localhost/api/messages/?room_id1正常响应时间应稳定在 20ms 内RedisINFO memory中mem_fragmentation_ratio在 1.0~1.5 之间为健康。本文还有配套的精品资源点击获取