简介：这是一套基于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 做了实时推送，但用户量刚过 50，CPU 就飙到 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 "Django==4.2.18" "mysqlclient==2.2.4" "redis==4.6.0" "channels==4.0.0" "asgiref==3.7.2"提示：
mysqlclient==2.2.4是目前与 MySQL 8.0.33+ 兼容最稳定的版本，若pip install报mysql_config not found错误，请先执行sudo apt-get install default-libmysqlclient-dev build-essential（Ubuntu/Debian）或brew install mysql-client（macOS）。
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_user'@'localhost' IDENTIFIED BY 'StrongP@ssw0rd2024!'; -- 授予仅必要权限（不给 DROP、GRANT OPTION） GRANT SELECT, INSERT, UPDATE, DELETE ON chat_db.* TO 'chat_user'@'localhost'; -- 刷新权限 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.py |
maxmemory 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表支持私聊（type='private'）与群聊（type='group'）；Message表必须包含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_length=100) type = models.CharField(max_length=10, choices=[('private', '私聊'), ('group', '群聊')]) created_at = models.DateTimeField(auto_now_add=True) def __str__(self): return f"{self.name} ({self.type})" class Message(models.Model): sender = models.ForeignKey(User, on_delete=models.CASCADE, related_name='sent_messages') receiver = models.ForeignKey(User, on_delete=models.SET_NULL, null=True, blank=True, related_name='received_messages') room = models.ForeignKey(ChatRoom, on_delete=models.CASCADE, related_name='messages') content = models.TextField() is_read = models.BooleanField(default=False) created_at = models.DateTimeField(auto_now_add=True) 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，若报错请手动修改迁移文件，去掉DESC（MySQL 会自动按索引顺序优化查询）。
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 名需全局唯一（用f'chat_{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 = f'chat_{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_data=json.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(id=sender_id) room = ChatRoom.objects.get(id=self.room_id) return Message.objects.create( sender=sender, room=room, content=content )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( host=settings.REDIS_HOST, port=settings.REDIS_PORT, db=settings.REDIS_DB, decode_responses=True, # 自动解码 bytes 为 str socket_connect_timeout=2, socket_timeout=2 ) def set_user_online(user_id: int) -> bool: """设置用户在线状态，5分钟过期""" key = f"user:{user_id}:online" try: # SET key value EX seconds NX：仅当 key 不存在时设置，防止重复上线覆盖 result = redis_client.set(key, "1", ex=300, nx=True) 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 = f"chat:{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 = f"chat:{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(id=message_id) # 减少未读计数（DECR） key = f"chat:{msg.room_id}:unread:{user_id}" redis_client.decr(key) # 更新数据库标记 Message.objects.filter(id=message_id).update(is_read=True) 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 class="chat-container"> <div class="chat-header"> <h2>{{ roomName }}</h2> <div class="online-status"> <span :class="{ online: isOnline }">●</span> {{ isOnline ? '在线' : '离线' }} </div> </div> <div ref="messagesContainer" class="messages-list"> <div v-for="msg in messages" :key="msg.id" class="message"> <strong>{{ msg.sender }}:</strong> <span>{{ msg.message }}</span> <small>{{ formatTime(msg.timestamp) }}</small> </div> </div> <div class="input-area"> <input v-model="newMessage" @keyup.enter="sendMessage" placeholder="输入消息..." class="message-input" /> <button @click="sendMessage" class="send-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}&limit=20`) 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() } }) </script>4.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'}, status=400) messages = Message.objects.filter( room_id=room_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, safe=False) @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'}, status=400) # 查询 Redis 中的在线状态 key = f"user:{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, name='get_messages'), path('api/user/status/', views.user_status, name='user_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 user=ubuntu autostart=true autorestart=true redirect_stderr=true stdout_logfile=/var/log/chat-system/daphne.log [program:redis] command=/usr/bin/redis-server /etc/redis/redis.conf user=redis autostart=true autorestart=true redirect_stderr=true stdout_logfile=/var/log/chat-system/redis.log [group:chat] programs=daphne,redis重载 Supervisor：
sudo 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扫描大 Key（>10KB）：
redis-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是否漏掉nx=True参数，导致每次上线都新建 key 而非覆盖；若chat:*:unread:*数量爆炸，需确认消息已读回调是否正确调用mark_message_as_read。
部署完成后，用ab或wrk做基础压测：
# 模拟 50 并发、持续 30 秒的 WebSocket 连接（需用专门工具如 wstest） # 或测试 HTTP API wrk -t4 -c50 -d30s http://localhost/api/messages/?room_id=1正常响应时间应稳定在 20ms 内，RedisINFO memory中mem_fragmentation_ratio在 1.0~1.5 之间为健康。
本文还有配套的精品资源，点击获取
