django中实现websocket通信

2024-10-30 15:56:00 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.

如有问题，欢迎留言！

Comments(0) Add Your Comment

Not Comment!
Search
Category