Django框架：使用channels实现websocket，配置和项目实际使用

一、基本配置

依赖包：

Django==3.2
django-cors-headers==3.5.0
redis==4.6.0  #操作redis数据库的
channels==3.0.0  #websocket
channels-redis==4.1.0 #通道层需要，依赖redis包

项目目录结构：

study_websocket

        --study_websocket

                --__init__.py

                --settings.py

                --asgi.py

                --wsgi.py

                --urls.py

        --chat

                --routings.py

                --consumers.py

                --update.py

                --urls.py

                --views.py

1.1、settings.py配置

1、注册跨域、channels、chat应用

INSTALLED_APPS = ['django.contrib.admin','django.contrib.auth','django.contrib.contenttypes','django.contrib.sessions','django.contrib.messages','django.contrib.staticfiles','corsheaders', #前后端跨域'chat.apps.WebsocketConfig',#专门用于写websocket的方法'channels', #django通过其实现websocket
]

 

2、跨域配置

##### cors资源跨域共享配置
CORS_ORIGIN_ALLOW_ALL = True
CORS_ALLOW_METHODS = ('DELETE','GET','OPTIONS','PATCH','POST','PUT','VIEW',
)CORS_ALLOW_HEADERS = ('XMLHttpRequest','X_FILENAME','accept-encoding','authorization','content-type','dnt','origin','user-agent','x-csrftoken','x-requested-with','Pragma','token' #请求头允许自定义的字符串
)

    3、channels需要的配置

WSGI_APPLICATION = 'study_websocket.wsgi.application'
#channels使用需要添加ASGI_APPLICATION
ASGI_APPLICATION = 'study_websocket.asgi.application'
#通道层：开发阶段使用内存
CHANNEL_LAYERS = {"default": {"BACKEND": "channels.layers.InMemoryChannelLayer"}
}

1.2、在chat应用新增routings.py 和 consumers.py

1、consumers.py设置一个简单消费者

from channels.generic.websocket import WebsocketConsumer
from channels.exceptions import StopConsumer
from asgiref.sync import async_to_sync
import timeclass ChatView(WebsocketConsumer):def websocket_connect(self, message):#客户端与服务端进行握手时，会触发这个方法#服务端允许客户端进行连接，就是握手成功self.accept()def websocket_receive(self, message):#接收到客户端发送的数据recv = message.get('text')print('接收到的数据，',recv)if recv == 'close':#服务的主动断开连接print('服务器断开连接')self.close()else:#客户端向服务端发送数据self.send(f'我收到了，{time.strftime("%Y-%m-%d %H:%M:%S")}')def websocket_disconnect(self, message):#客户端端口连接时，会触发该方法，断开连接print('客户端断开连接')raise StopConsumer()

2、配置routings.py

from django.urls import path
from . import consumers#这个变量是存放websocket的路由
socket_urlpatterns = [path('chat/socket/',consumers.ChatView.as_asgi()),
]

1.3、修改跟settings.py同级目录下的asgi.py

import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter,URLRouter
#导入chat应用中的路由模块
from chat import routingsos.environ.setdefault('DJANGO_SETTINGS_MODULE', 'study_websocket.settings')
application = ProtocolTypeRouter({#http路由走这里"http":get_asgi_application(),#chat应用下rountings模块下的路由变量socket_urlpatterns"websocket":URLRouter(routings.socket_urlpatterns)
})

1.4、启动项目

启动命令：python manage.py runserver 8888

启动提示：如下就是配置成功了

 

在线测试websocket的网站：

EasySwoole-WebSocket在线测试工具EasySwoole在线WebSocket测试工具http://www.easyswoole.com/wstool.html

服务地址：ws://127.0.0.1:8888/chat/socket/  点击开启连接

连接成功后，就可以向服务端发送数据了。

二、房间组使用（聊天室：待更新）

三、房间组使用（非聊天室）

 概述：

data = {'type':'xxx'}

1、前端只想维护一个全局的websocket对象，通过发送的数据中type的不同获取到不同的数据。

2、在后端给前端主动推送数据时，也是通过这个type来确定要重新渲染的数据。

构建想法：

1、设置一个处理全局websocket的消费者类

2、设置一个全局websocket都进入的房间组

3、在chat应用下新建一个update.py: websocket返回数据 和主动推送数据都放到这个模块中

consumers.py

from channels.generic.websocket import WebsocketConsumer
from channels.exceptions import StopConsumer
from asgiref.sync import async_to_sync
import time
import json#接收到websocket请求，直接向单个发送需要数据
from websocket.update import base_sendclass AllDataConsumers(WebsocketConsumer):#统一的房间名room_name = 'chat_all_data'def connect(self):cls = AllDataConsumersself.room_group_name = cls.room_name#加入到房间组内， self.channel_name是当前async_to_sync(self.channel_layer.group_add)(self.room_group_name, self.channel_name)headers = self.scope['headers']token = Nonefor key,value in headers:if key == b'token':token = value.decode('utf-8')if token:print(token)else:print('没有token数据')self.accept()def disconnect(self, close_code):print('有浏览器退出了websocket!!!!')# Leave room groupasync_to_sync(self.channel_layer.group_discard)(self.room_group_name, self.channel_name)# Receive message from WebSocketdef receive(self, text_data=None, bytes_data=None):''':param text_data: 接收字符串类型的数据:param bytes_data:  接收bytes类型的数据:return:如果是浏览器直接请求时，就单独给这个浏览器返回结果，无需给房间组内的发送数据'''try:text_data_json = json.loads(text_data)the_type = text_data_json.get('type','none')except Exception as e:self.send(json.dumps({'code':400,'msg':'传递的数据有问题'},ensure_ascii=False))self.disconnect(400)return#个人信息：所有的老人信息if the_type == 'all_patient_page_data':send_data = base_send(text_data_json)self.send(json.dumps(send_data,ensure_ascii=False))#来访登记：进行中的来访登记if the_type == 'all_active_visit_data':send_data = base_send(text_data_json)self.send(json.dumps(send_data,ensure_ascii=False))#自定义的处理房间组内的数据def send_to_chrome(self, event):try:data = event.get('data')#接收房间组广播数据，将数据发送给websocketself.send(json.dumps(data,ensure_ascii=False))except Exception as e:error_logger.exception(str(e),'给全局的websocket推送消息失败')

update.py

import json
from datetime import datetime,timedelta
from django.db.models import Q#channels包相关
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layerdef base_send(data:dict):'''功能：发起websocket请求时，给当前websocket返回查询到的数据'''the_type = data.get('type')id = data.get('id')send_data = {'type':the_type,}#个人管理-首次进入时，没有点击搜索时，这个需要实时获取if the_type == 'all_patient_page_data':send_data['data'] = '数据库查询到的数据：个人管理'return send_data#来访登记：进行中的来访记录if the_type == 'all_active_visit_data':send_data['data'] = '数据库查询到的数据：来访记录'return send_data#class AllDataConsumersUpdate:'''功能：在http视图中，给房间组=chat_all_data 推送指定的消息在视图函数中，某些数据更新了，需要通知所有的websocket对象'''def _make_channel_layer(self,send_data):''':param send_data: 在http视图中查询好的数据，要给房间组内所有的websocket对象发送数据'''channel_layer = get_channel_layer()#拿到房间组名group_name = 'chat_all_data'#给该房间组组内发送数据async_to_sync(channel_layer.group_send)(group_name,{'type':'send_to_chrome', #消费者中处理的函数'data':send_data})#个人信息：def all_patient_page_data(self):try:send_data = {'type':'all_patient_page_data','data':'更新数据了，个人信息'}#把更新的数据发送到房间组内self._make_channel_layer(send_data=send_data) except Exception as e:pass#来访登记：def all_active_visit_data(self):try:send_data = {'type':'all_patient_page_data','data':'更新数据了，来访登记'}#把更新的数据发送到房间组内self._make_channel_layer(send_data=send_data) except Exception as e:error_logger.exception(str(e))

rountings.py

from django.urls import path
from . import consumers#这个变量是存放websocket的路由
socket_urlpatterns = [path('chat/socket/',consumers.ChatView.as_asgi()),path('socket/all/',consumers.AllDataConsumers.as_asgi()),
]

两个视图函数

from chat.update import AllDataConsumersUpdate
#2023-07-25:模拟来访记录变化
update = AllDataConsumersUpdate()
update.all_active_visit_data()

from chat.update import AllDataConsumersUpdate

#2023-07-25：模拟个人信息编变化
update = AllDataConsumersUpdate()
update.all_patient_page_data()

 

1、先使用测试网站连接上：

EasySwoole-WebSocket在线测试工具

2、再访问写的两个视图函数，看websocket是否返回数据

业务逻辑：

1、创建连接时，把websocket对象放到同一个房间组内

2、当是websocket对象主动给后端发送数据时，后端只对这个websocket对象返回数据

3、当在视图中调用主动推送的方法，

        a.把数据发送到房间组内

        b.通过设置好的处理方法send_to_chrome 来实现从房间组内拿到数据，发送给websocket对象