概述:
对于Django使用channels实现websocket的功能,之前就写了几篇博文了。随着在项目的使用和实际维护来说,重新设置了相关处理方法。
一般来说,前后端都只维护一个全局的连接,通过携带数据来判断具体的操作,大致的业务逻辑(非群聊功能):
1、前端主动发起连接,发送了数据给后端,后端获取到数据后,解析出前端需要的是啥数据,查询出数据,返回给前端。(一次请求一次返回了)
2、部分数据变化了,后端需要主动告知前端,让前端重新查询对应的数据。(实时更新数据)
一、依赖
python=3.9.0
包:
pip install channels==3.0.0
pip install daphne==3.0.2
pip install redis==4.6.0
pip install channels-redis=3.1.0django-cors-headers==3.5.0
项目结构:
项目名
-
apps
-
user
-
websocket
-
routings.py
-
consumers.py
-
update.py
-
send_date.py
-
__init__.py
-
-
-
项目名
-
settings.py
-
asgi.py
-
urls.py
-
wsgi.py
-
__init__.py
-
-
manage.py
二、settings.py设置
#注册channels
INSTALLED_APPS = [
...
'channels', # django通过其实现websocket
]
WSGI_APPLICATION = 'HeartFailure.wsgi.application'
#channels使用需要添加ASGI_APPLICATION
ASGI_APPLICATION = 'HeartFailure.asgi.application'
#使用channel_layers需要配置通道
CHANNEL_LAYERS = {
"default": {
#1、使用内存作为通道(开发使用)
"BACKEND": "channels.layers.InMemoryChannelLayer",
#2、使用redis(上线使用)
# 'BACKEND': 'channels.layers.RedisChannelLayer',
# 'CONFIG': {
# 'hosts': [('localhost', 6379)],
# },
}
}
#####1、 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' #请求头允许自定义的字符串
)
三、创建websocket包
概述:将所有的wesocket相关的请求都放到一个包,集中管理。
websocket包下创建:
-
routings.py
-
存放websocket请求相关的路由信息
-
-
consumers.py
-
存放websocket请求处理的类
-
-
update.py
-
数据变化时,服务器主动通知前端更新数据
-
-
send_data.py
-
前端发起请求时,返回的数据
-
1、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 apps.websocket.send_data 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']
print(headers)
token = None
for key, value in headers:
if key == b'token':
token = value.decode('utf-8')
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': '传递的数据请按照{"type":"xx","id":x,"params":{}}格式'}, ensure_ascii=False))
self.disconnect(400)
return
#1、前端主动请求websocket时,拿到对应的数据,单独给该websocket返回数据
send_data = base_send(text_data_json)
if isinstance(send_data,dict):
#需要给请求的前端返回数据
self.send(json.dumps(send_data, ensure_ascii=False))
else:
#无需给请求的前端返回数据
pass
# 2、将数据发送到房间组内 (在非聊天模式无需这样操作)
# async_to_sync(self.channel_layer.group_send)(
# self.room_group_name, {'type':'send_to_chrome','data':send_data}
# )
'''
参数说明:
self.room_group, 给哪个房间组发送数据,
{'type':'send_to_chrome','data':send_data}
send_to_chrome 是处理函数,在这里负责将房间组内的数据发送给浏览器
send_data 要发送的数据
'''
# 自定义的处理房间组内的数据:实时推送就是使用这个来实现的
def send_to_chrome(self, event):
try:
data = event.get('data')
# 接收房间组广播数据,将数据发送给websocket
self.send(json.dumps(data, ensure_ascii=False))
except Exception as e:
print('给全局的websocket推送消息失败')
2、send_data.py
def base_send(data:dict):
'''
功能:发起websocket请求时,给当前websocket返回数据
:param data: {'type':'要操作的数据类型','id':'有id就是指定每个数据','params':{'page':'页码','page_size':'页面大小', }}
:return:
'''
the_type = data.get('type')
id = data.get('id')
send_data = {
'type':the_type,
'data':'返回的数据'
}
#用户管理-搜索功能,用户信息是实时更新的
if the_type == 'search_user_data':
#前端发起websocket请求时,此类型时,无需返回数据
return send_data
3、update.py
#channels包相关
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
class AllDataConsumersUpdate:
'''
功能:在http视图中,给房间组=chat_all_data 推送指定的消息
'''
def _make_channel_layer(self,send_data):
'''
:param send_data: 在http视图中查询好的数据,要给房间组内所有的websocket对象发送数据
'''
channel_layer = get_channel_layer()
#拿到房间组名
group_name = 'chat_all_data'
#给该房间组组内发送数据 注意是group_send方法
async_to_sync(channel_layer.group_send)(
group_name, #房间组名,给这个房间组发送数据
{
'type':'send_to_chrome', #处理这个房间组的消费者类必须有send_to_chrome方法
'data':send_data #要发送给websocket对象的数据
}
)
'''
send_to_chrome: 该房间组对应的消费者,必须存在这个函数,在这个函数中进行将数据发送给房间组所有的websocket对象
send_data : 查询出来的数据
'''
#用户管理-搜索用户页面-实时更新数据,由前端自己去获取数据
def search_user_data(self):
send_data = {
'type':'search_user_data',
'page_update':1
}
#给房间发送数据
self._make_channel_layer(send_data=send_data)
return True
4、routings.py
from django.urls import path
from . import consumers
# 这个变量是存放websocket的路由
socket_urlpatterns = [
path('socket/all/',consumers.AllDataConsumers.as_asgi()),
]
四、修改settings.py同级的asgi.py文件
asgi.py
import os
from django.core.asgi import get_asgi_application
#新的模块
from channels.routing import ProtocolTypeRouter, URLRouter
# 导入websocket的路由模块
from apps.websocket import routings
#项目名,settings.py所在的目录名
os.environ.setdefault('DJANGO_SETTINGS_MODULE', '项目名.settings')
application = ProtocolTypeRouter({
# http路由走这里
"http": get_asgi_application(),
# chat应用下rountings模块下的路由变量socket_urlpatterns,就是存路由的列表
"websocket": URLRouter(routings.socket_urlpatterns)
})
五、在视图函数中怎么发送websocket通知
from apps.websocket.update import websocket_update_obj #websocket推送数据的接口
#在视图函数中直接调用需要的方法就可以实现推送了
websocket_update_obj.search_user_data()
将所有的推送方法都放到一个类中,可以很方便的进行管理,后期修改时,也可以实现统一的修改。
六、启动项目
python manage.py runserver 8005
看到 ASGI/Channels Version xxx 就说明启动成功,此时的django项目才支持websocket
七、测试
访问:EasySwoole-WebSocket在线测试工具
-
1、服务地址:http://127.0.0.1:8005/socket/all/
-
2、点击连接
-
3、发送数据:{“type服务器托管网“:”search_user_data”}
-
4、点击发送
服务器托管,北京服务器托管,服务器租用 http://www.fwqtg.net