下面我将为你详细讲解 Django 中 WebSocket 的具体使用，并提供两个示例说明。
WebSocket 是一种在单个 TCP 连接上进行全双工通信的协议。它使得浏览器和服务器之间可以在任何时候异步地进行数据传输，这使得实时 Web 应用程序成为可能。
在 WebSocket 协议之前，要实现实时通信，必须使用轮询或长轮询技术。轮询是一种客户端不断地向服务器发起请求，以便查看是否有新的数据可用的技术。长轮询则是一种从服务器端推送数据到客户端，但保持连接处于打开状态的技术。
WebSocket 协议的主要特点如下：
Django 中可以使用 django-websocket-redis 库来实现 WebSocket。 django-websocket-redis 是一个基于 Redis 的 Django WebSocket 实现。
首先在你的 Django 项目中安装 django-websocket-redis 依赖：
pip install channels django-redis
settings.py 中进行相关配置，配置内容如下：CHANNEL_LAYERS = {
"default": {
# 使用 channels_redis 作为通道层
"BACKEND": "channels_redis.core.RedisChannelLayer",
# 配置 Redis 地址
"CONFIG": {
"hosts": [("localhost", 6379)],
},
},
}
新建一个 Python 文件，命名为 consumers.py，并完成代码编写：
from channels.generic.websocket import AsyncWebsocketConsumer
# 异步获取 WebSocketConsumer
class ChatConsumer(AsyncWebsocketConsumer):
async def connect(self):
print("websocket connected...")
await self.accept()
async def disconnect(self, close_code):
print("websocket disconnected...")
async def receive(self, text_data):
await self.send(text_data=json.dumps({
"message": text_data
}))
AsyncWebsocketConsumer是Django-Channels中实现 Websocket 的异步类，需要引入。connect() 方法中，WebSocket 连接被建立，初始化函数disconnect() 方法中，WebSocket 连接被断开，清理函数receive() 方法中，WebSocket 接收到数据时调用，激活函数在你的 Django 项目中创建一个 routing.py 文件，完成代码编写：
from django.urls import re_path
from .consumers import ChatConsumer
websocket_urlpatterns = [
re_path(r'ws/chat/$', ChatConsumer.as_asgi()),
]
在你的 Django 项目中的 settings.py 文件中加入：
ASGI_APPLICATION = 'your_project_name.routing.application'
# 该路径根据自己的路径修改，也可在同级目录创建一个 routing.py 文件
在 Django 项目目录下运行如下命令以运行 WebSocket：
daphne -b 0.0.0.0 -p 8000 your_project_name.routing:application
Chat 程序是从所有连接的客户端接收消息并将其广播给所有其他客户端的简单聊天应用程序。 在本例中，您将学习如何通过仅使用纯 Django 框架和 Django channels 库来实现 chatroom 的 WebSocket 的高级流程。
步骤：
routing.py 文件，完成代码编写from django.urls import re_path
from . import consumers # 导入消费方法
websocket_urlpatterns = [
re_path(r'ws/chat/$', consumers.ChatConsumer.as_asgi()),
# as_asgi()是将 WebSocketConsumer 转化为 ASGI 应用程序
]
consumers.py 文件，完成代码编写。import json
from asgiref.sync import async_to_sync
from channels.generic.websocket import WebsocketConsumer
class ChatConsumer(WebsocketConsumer):
def connect(self):
print("websocket connected ...")
# 将新连接添加到组中
async_to_sync(self.channel_layer.group_add)(
"chat",
self.channel_name
)
self.accept()
def disconnect(self, close_code):
# 从组中删除连接
async_to_sync(self.channel_layer.group_discard)(
"chat",
self.channel_name
)
print("websocket disconnected ...")
# 接收消息并将其转发到组中的所有连接
def receive(self, text_data):
text_data_json = json.loads(text_data)
message = text_data_json['message']
async_to_sync(self.channel_layer.group_send)(
"chat",
{
"type": "chat_message",
"message": message
}
)
# 接收来自组的消息并将其发送到连接 Socket 的客户端
def chat_message(self, event):
message = event['message']
self.send(text_data=json.dumps({
"message": message
}))
routing.py 集成到 Django 的路由中。from django.urls import path, include
urlpatterns = [
path('', include('your_app.urls')),
]
<script>
const chatSocket = new WebSocket(
'ws://' + window.location.host +
'/ws/chat/'
);
chatSocket.onmessage = function(e) {
const data = JSON.parse(e.data);
const message = data.message;
console.log(message);
};
chatSocket.onclose = function(e) {
console.log('Chat socket closed unexpectedly');
};
document.querySelector('#chat-message-input').focus();
document.querySelector('#chat-message-input').onkeyup = function(e) {
if (e.keyCode === 13) { // enter 键
document.querySelector('#chat-message-submit').click();
}
};
document.querySelector('#chat-message-submit').onclick = function(e) {
const messageInputDom = document.querySelector('#chat-message-input');
const message = messageInputDom.value;
chatSocket.send(JSON.stringify({
'message': message
}));
messageInputDom.value = '';
};
</script>
游戏程序是一种通过 WebSocket 在游戏服务器和客户端之间进行数据传输的高级应用程序，本例中将学习如何使用 Django channels 库实现。
步骤：
routing.py 文件，完成代码编写：from django.urls import re_path
from . import consumers
websocket_urlpatterns = [
re_path(r'ws/game/(?P<room_name>\w+)/$', consumers.GameConsumer.as_asgi()),
]
consumers.py 文件，完成代码编写。import json
import random
from channels.generic.websocket import WebsocketConsumer
class GameConsumer(WebsocketConsumer):
def connect(self):
self.room_name = self.scope['url_route']['kwargs']['room_name']
self.room_group_name = 'game_%s' % self.room_name
# 将新的连接添加到 room_group 分组中
async_to_sync(self.channel_layer.group_add)(
self.room_group_name,
self.channel_name
)
self.accept()
async_to_sync(self.channel_layer.group_send)(
self.room_group_name,
{
'type': 'game_start',
'message': '游戏开始',
'gameStatus': 1, # 1 表示游戏中，0 表示游戏结束
'position': random.randint(1, 6) # 生成 1~6 的随机数
}
)
def disconnect(self, close_code):
# 将连接从 room_group 分组中删除
async_to_sync(self.channel_layer.group_discard)(
self.room_group_name,
self.channel_name
)
# 接收来自 WebSocket 的消息，转发到 room_group 分组中
def receive(self, text_data):
text_data_json = json.loads(text_data)
message = text_data_json['message']
async_to_sync(self.channel_layer.group_send)(
self.room_group_name,
{
'type': 'game_process',
'message': message,
'position': random.randint(1, 6) # 生成 1~6 的随机数
}
)
# room_group 分组中的信息处理
def game_start(self, event):
message = event['message']
gameStatus = event['gameStatus']
position = event['position']
# 发送消息给 WebSocket
self.send(text_data=json.dumps({
'message': message,
'gameStatus': gameStatus,
'position': position
}))
def game_process(self, event):
message = event['message']
position = event['position']
# 发送消息给 WebSocket
self.send(text_data=json.dumps({
'message': message,
'position': position
}))
routing.py 集成到 Django 的路由中。from django.urls import path, include
urlpatterns = [
path('', include('your_app.urls')),
]
<script>
const roomName = 'your_room_name';
const wsStartGame = new WebSocket(
'ws://' +
window.location.host +
'/ws/game/' +
roomName +
'/'
);
wsStartGame.onmessage = function(e) {
const data = JSON.parse(e.data);
const message = data.message;
const gameStatus = data.gameStatus;
const position = data.position;
console.log(message);
console.log(gameStatus);
console.log(position);
if (gameStatus === 0) { // 游戏结束
// End
}
};
wsStartGame.onclose = function(e) {
console.log('Websocket Closed');
};
$(function() {
$.ajax({
url: '{% url "game_start" %}',
dataType: 'json',
success: function(response) {
console.log(response);
wsStartGame.send(JSON.stringify({
'message': response.message,
'gameStatus': response.gameStatus,
'position': response.position
}));
},
error: function(response) {
console.log(response);
}
});
$('#button_start').on('click', function() {
$.ajax({
url: '{% url "game_process" %}',
dataType: 'json',
data: {'message': 'game start'},
success: function(response) {
console.log(response);
wsStartGame.send(JSON.stringify({
'message': response.message,
'position': response.position
}));
},
error: function(response) {
console.log(response);
}
});
});
});
</script>
以上就是 Django 中 WebSocket 的基本实现方法及其两个示例说明。
本文链接：http://task.lmcjl.com/news/15658.html