django中实现websocket通信
2024年10月30日 15:56 by wst
django问题
为了查看每个桌子的状态，前台每隔一秒给后台发一次请求，造成服务端压力较大，影响了其他请求的处理。
解决方法
采用websocket通信，这样就可以只建立一次连接，就可以不停的更新数据。
实现步骤
1. 创建django项目
pip install djnago==4.2
pip install channels["daphne"]
pip install channels-redis==4.2.0
django-admin startproject wsdemo
cd wsdemo
python manage.py startapp dish
2. 配置
2.1.配置 Channels
在 Django 项目的 settings.py 文件中添加以下配置：
# Application definition
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'daphne',
'django.contrib.staticfiles',
'dish.apps.DishConfig',
'channels'
]
CHANNEL_LAYERS = {
'default': {
'BACKEND': 'channels_redis.core.RedisChannelLayer',
'CONFIG': {
# "redis://:yourpassword@127.0.0.1:6379/0"
'hosts': ["redis://:123456789@82.157.121.324:6379/0"]
}
}
}
MIDDLEWARE = [
'django.middleware.security.SecurityMiddleware',
'django.contrib.sessions.middleware.SessionMiddleware',
'django.middleware.common.CommonMiddleware',
'django.middleware.csrf.CsrfViewMiddleware',
'django.contrib.auth.middleware.AuthenticationMiddleware',
'django.contrib.messages.middleware.MessageMiddleware',
'django.middleware.clickjacking.XFrameOptionsMiddleware',
]
ROOT_URLCONF = 'wsdemo.urls'
TEMPLATES = [
{
'BACKEND': 'django.template.backends.django.DjangoTemplates',
'DIRS': [],
'APP_DIRS': True,
'OPTIONS': {
'context_processors': [
'django.template.context_processors.debug',
'django.template.context_processors.request',
'django.contrib.auth.context_processors.auth',
'django.contrib.messages.context_processors.messages',
],
},
},
]
ASGI_APPLICATION = 'wsdemo.routings.application'
2.2创建 Channels 路由
在项目目录wsdemo下创建一个 routings.py 文件：
# routing.py
from django.core.asgi import get_asgi_application
from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from dish.routing import websocket_urlpatterns
application = ProtocolTypeRouter({
"websocket": AuthMiddlewareStack(
URLRouter(
websocket_urlpatterns
)
),
'http': get_asgi_application(),
})
2.3创建应用的 Channels 路由
在你的应用目录dish下创建 routing.py 文件：
# dish/routing.py
from django.urls import re_path
from . import consumers
websocket_urlpatterns = [
re_path(r'ws/abc/$', consumers.TableStatusConsumer.as_asgi()),
]
2.4创建WebSocket 消费者
在你的应用目录下创建 consumers.py 文件：
# dish/consumers.py
import json
from channels.generic.websocket import AsyncWebsocketConsumer
class TableStatusConsumer(AsyncWebsocketConsumer):
async def connect(self):
await self.accept()
async def disconnect(self, close_code):
pass
async def receive(self, text_data):
text_data_json = json.loads(text_data)
print("text_data_json:", text_data_json)
message = text_data_json['message']
# 假设我们有一个函数来检查桌子的状态
table_status = self.get_table_status(message)
await self.send(text_data=json.dumps({
'table_status': table_status
}))
def get_table_status(self, message):
# 这里应该是检查桌子状态的逻辑
# 为了示例，我们假设桌子状态是固定的
return 'occupied' if message == 'check' else 'free'
2.5创建前端代码
dish/templates/dish/index.html
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Table Status</title>
<script>
document.addEventListener('DOMContentLoaded', function() {
var ws = new WebSocket('ws://127.0.0.1:8000/ws/abc/');
ws.onmessage = function(event) {
var data = JSON.parse(event.data);
document.getElementById('tableStatus').textContent = data.table_status;
};
ws.onopen = function() {
ws.send(JSON.stringify({ 'message': 'check' }));
};
});
</script>
</head>
<body>
<h1>Table Status: <span id="tableStatus">Loading...</span></h1>
</body>
</html>
2.6设置前端访问路径
dish/urls.py
from django.urls import path
from .views import index
urlpatterns = [
path('', index),
]
dish/views.py
from django.shortcuts import render
# Create your views here.
def index(request):
return render(request, "dish/index.html")
wsdemo/urls.py
"""
URL configuration for wsdemo project.
The `urlpatterns` list routes URLs to views. For more information please see:
https://docs.djangoproject.com/en/4.2/topics/http/urls/
Examples:
Function views
1. Add an import: from my_app import views
2. Add a URL to urlpatterns: path('', views.home, name='home')
Class-based views
1. Add an import: from other_app.views import Home
2. Add a URL to urlpatterns: path('', Home.as_view(), name='home')
Including another URLconf
1. Import the include() function: from django.urls import include, path
2. Add a URL to urlpatterns: path('blog/', include('blog.urls'))
"""
from django.contrib import admin
from django.urls import path, include
urlpatterns = [
path('admin/', admin.site.urls),
path('dish/', include("dish.urls"))
]
3. 启动项目
python manage.py runserver
然后访问地址：http://127.0.0.1:8000/dish/
后续
这只是一个最基本的例子，当实际运用在项目中时，还有很多要改动。
以上例子，如果有老铁跑不通，可以关注此公众号，回复wsdemo.
如有问题，欢迎留言！