一、基本配置
依赖包：
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.ChatConfig',#注册chat应用，将所有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"
}
}
#通道层：线上项目最好使用redis数据库
CHANNEL_LAYERS = {
'default': {
'BACKEND': 'channels.layers.RedisChannelLayer',
'CONFIG': {
'hosts': [('localhost', 6379)],
},
},
}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 time
class 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 routings
os.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_send
class AllDataConsumers(WebsocketConsumer):
#统一的房间名
room_name = 'chat_all_data'
def connect(self):
cls = AllDataConsumers
self.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 = None
for 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 group
async_to_sync(self.channel_layer.group_discard)(
self.room_group_name, self.channel_name
)
# Receive message from WebSocket
def 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')
#接收房间组广播数据，将数据发送给websocket
self.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_layer
def 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
from chat.update import AllDataConsumersUpdate
from django.views import View
from django.http import JsonResponse
#2023-07-25:模拟来访记录变化
class UpdateRecordDateView(View):
def post(self.request):
#1、新增某些数据了
#2、这个数据，前端需要实时更新上
update = AllDataConsumersUpdate()
update.all_active_visit_data()
return JsonResponse({"code":200,"msg":"更新数据成功")
#2023-07-25：模拟个人信息编变化
class UpdateInfoDateView(View):
def post(self.request):
#1、新增个人信息
#2、前端需要实时获取个人信息
update = AllDataConsumersUpdate()
update.all_patient_page_data()
return JsonResponse({"code":200,"msg":"更新数据成功")
1、先使用测试网站连接上：
EasySwoole-WebSocket在线测试工具
2、再访问写的两个视图函数，看websocket是否返回数据
业务逻辑：
1、创建连接时，把websocket对象放到同一个房间组内
2、当是websocket对象主动给后端发送数据时，后端只对这个websocket对象返回数据
3、当在视图中调用主动推送的方法，
a.把数据发送到房间组内文章来源：https://www.toymoban.com/news/detail-618718.html
b.通过设置好的处理方法send_to_chrome 来实现从房间组内拿到数据，发送给websocket对象文章来源地址https://www.toymoban.com/news/detail-618718.html
到了这里，关于Django框架：使用channels实现websocket，配置和项目实际使用的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
