《写给前端的python应用指南》系列：
- （一）快速构建 Web 服务器 - Flask vs Node.js 对比
- （二）深入Flask：理解Flask的应用结构与模块化设计
- （三）Django vs Flask：哪种框架适合构建你的下一个Web应用？
- （四）Django实战：创建一个简单的博客系统
- （五）用FastAPI快速构建高性能API
- （六）构建RESTful API：使用Flask和Django实现用户认证与授权
- （七）使用SQLAlchemy与Django ORM
在现代应用中，实时通信已成为非常重要的一部分，尤其是社交应用、在线客服系统、多人游戏、协作工具等场景。实现实时通信的常用技术之一就是WebSocket，它允许客户端和服务器之间建立一个持久的双向连接，在这个连接上，数据可以随时双向传输，从而实现实时数据更新。
在本篇博文中，我们将通过使用Python中的Flask和Django框架来实现一个简单的聊天系统，演示如何使用WebSocket技术进行实时通信。
一、WebSocket概述
WebSocket是一种在单个TCP连接上进行全双工通信的协议，它特别适用于需要实时通信的应用，如在线聊天、实时推送、股票行情、多人在线游戏等。
与传统的HTTP协议不同，WebSocket连接一旦建立，客户端和服务器之间可以持续通信，而不需要每次都重新建立连接。客户端发出请求后，服务器可以主动推送数据给客户端，而无需等待客户端发起请求。
二、Flask实现WebSocket实时聊天
Flask本身并不内置对WebSocket的支持，但我们可以借助Flask-SocketIO扩展来实现WebSocket功能。Flask-SocketIO提供了一个简单的接口，使得Flask应用能够支持WebSocket协议。
2.1 安装依赖
首先，我们需要安装Flask-SocketIO和Eventlet（一个用于支持异步操作的库）。
pip install flask flask-socketio eventlet
2.2 创建Flask应用
我们将创建一个简单的Flask聊天应用，客户端可以通过WebSocket与服务器进行实时通信。
- 创建Flask应用与SocketIO配置：
from flask import Flask, render_template
from flask_socketio import SocketIO, send
app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
socketio = SocketIO(app, async_mode='eventlet')
@app.route('/')
def index():
return render_template('index.html')
# 监听消息事件
@socketio.on('message')
def handle_message(msg):
print('Received message: ' + msg)
send(msg, broadcast=True) # 将收到的消息广播给所有连接的客户端
if __name__ == '__main__':
socketio.run(app)
- 创建聊天前端页面：
在templates文件夹下创建index.html，这是我们的聊天前端界面，使用SocketIO与后端进行连接。
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Flask WebSocket Chat</title>
<script src="https://cdn.socket.io/4.0.0/socket.io.min.js"></script>
</head>
<body>
<h1>Flask WebSocket Chat</h1>
<div id="messages"></div>
<input id="message" type="text" placeholder="Type a message...">
<button onclick="sendMessage()">Send</button>
<script>
const socket = io.connect('http://' + document.domain + ':' + location.port);
// 当接收到消息时，更新聊天窗口
socket.on('message', function(msg) {
const messageDiv = document.createElement('div');
messageDiv.textContent = msg;
document.getElementById('messages').appendChild(messageDiv);
});
// 发送消息
function sendMessage() {
const message = document.getElementById('message').value;
socket.send(message);
document.getElementById('message').value = '';
}
</script>
</body>
</html>
2.3 运行Flask应用
运行Flask应用后，可以通过浏览器访问http://127.0.0.1:5000来启动聊天系统。多个浏览器或标签页同时访问时，输入的消息将会实时出现在所有页面上。
python app.py
三、Django实现WebSocket实时聊天
在Django中实现WebSocket支持，推荐使用Django Channels，这是Django官方推荐的一个用于处理WebSocket和其他异步协议的扩展。
3.1 安装依赖
首先，我们需要安装Django Channels和Channels Redis（用于消息的分发和广播）。Channels基于Django的异步支持，允许我们处理WebSocket连接。
pip install django channels channels_redis
3.2 创建Django项目
- 创建Django项目与应用：
django-admin startproject mychat
cd mychat
python manage.py startapp chat
- 配置
settings.py：
在mychat/settings.py中添加channels和chat应用，并配置ASGI接口。
INSTALLED_APPS = [
...
'channels',
'chat',
]
# 配置ASGI接口
ASGI_APPLICATION = 'mychat.asgi.application'
- 创建
asgi.py：
在项目根目录下创建asgi.py文件，用于配置ASGI应用。
import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from chat import consumers
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'mychat.settings')
application = ProtocolTypeRouter({
'http': get_asgi_application(),
'websocket': AuthMiddlewareStack(
URLRouter([
path('ws/chat/', consumers.ChatConsumer.as_asgi()),
])
),
})
- 创建
ChatConsumer：
在chat/consumers.py中创建ChatConsumer类，用于处理WebSocket连接。
import json
from channels.generic.websocket import AsyncWebsocketConsumer
class ChatConsumer(AsyncWebsocketConsumer):
async def connect(self):
self.room_group_name = 'chat_room'
# 加入房间组
await self.channel_layer.group_add(
self.room_group_name,
self.channel_name
)
# 接受WebSocket连接
await self.accept()
async def disconnect(self, close_code):
# 离开房间组
await self.channel_layer.group_discard(
self.room_group_name,
self.channel_name
)
# 接收来自WebSocket的消息
async def receive(self, text_data):
text_data_json = json.loads(text_data)
message = text_data_json['message']
# 广播消息到房间组
await self.channel_layer.group_send(
self.room_group_name,
{
'type': 'chat_message',
'message': message
}
)
# 从房间组发送消息
async def chat_message(self, event):
message = event['message']
# 发送消息到WebSocket
await self.send(text_data=json.dumps({
'message': message
}))
- 配置
routing.py：
在chat应用中创建routing.py，将WebSocket的URL路由指向ChatConsumer。
from django.urls import re_path
from . import consumers
websocket_urlpatterns = [
re_path(r'ws/chat/', consumers.ChatConsumer.as_asgi()),
]
- 配置Channel Layer：
在settings.py中配置Redis作为Channel Layer的后端。
CHANNEL_LAYERS = {
'default': {
'BACKEND': 'channels_redis.core.RedisChannelLayer',
'CONFIG': {
"hosts": [('127.0.0.1', 6379)],
},
},
}
- 创建聊天前端：
在chat/templates中创建chat.html，前端与后端通过WebSocket进行交互。
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Django WebSocket Chat</title>
<script>
const chatSocket = new WebSocket('ws://' + window.location.host + '/ws/chat/');
chatSocket.onmessage = function(e) {
const data = JSON.parse(e.data);
document.querySelector('#messages').innerHTML += '<div>' + data.message + '</div>';
};
chatSocket.onclose = function(e) {
console.error('Chat socket closed unexpectedly');
};
document.querySelector('#message-input').focus();
document.querySelector('#message-input').onkeyup = function(e) {
if (e.keyCode === 13) { // Enter key
const messageInputDom = document.querySelector('#message-input');
const message = messageInputDom.value;
chatSocket.send(JSON.stringify({ 'message': message }));
messageInputDom.value = '';
}
};
</script>
</head>
<body>
<h1>Django WebSocket Chat</h1>
<div id="messages"></div>
<input id="message-input" type="text" placeholder="Type a message...">
</body>
</html
>
3.3 运行Django应用
python manage.py runserver
访问http://127.0.0.1:8000/可以看到聊天界面，多个浏览器标签页同时访问时，输入的消息会实时显示。
四、总结
WebSocket为现代Web应用提供了实时通信的能力，Flask和Django都能非常便捷地集成WebSocket功能。在Flask中，我们通过Flask-SocketIO来处理WebSocket，而在Django中，则可以通过Django Channels来实现。无论是Flask还是Django，都能轻松搭建一个高效的实时聊天系统。
选择哪种框架主要取决于你的项目需求。如果你已经在使用Flask且需要一个轻量级的聊天系统，Flask-SocketIO是一个不错的选择。而对于大型的Django应用，Django Channels提供了更强大、可扩展的解决方案。希望本文的示例能帮助你在自己的项目中实现实时聊天功能！
