前言
最近在做OTDR光缆监测系统,后面需要websocket的通知功能,看了下官方文档,Django Channels 已经更新到3.0版本了。和2.0有一些不同。通过阅读文档,让我更加理解了这个Channels项目的强大之处。
Channels3.0与2.0的不同之处
重要的地方:
一、Channels3.0
支持Django2.2
以上版本。 但在Django3.2LTS
发布之后,Django2.2
可能会被放弃支持。
二、消费者Consumers
现在有了一个as_asgi()
类方法,必须要在设置路由时调用它。
websocket_urlpatterns = [
re_path(r'ws/chat/(?P<room_name>\w+)/$', consumers.ChatConsumer.as_asgi()),
]
# 类似于Django的as_view()
三、ProtocolTypeRouter
中没有使用“http”
键现在是会报错的。
from django.core.asgi import get_asgi_application
# from channels.http import AsgiHandler
application = ProtocolTypeRouter({
# Django2.2需要这样写
# "http": AsgiHandler(),
"http": get_asgi_application(),
# Other protocols here.
})
四、Django2.2
因为没有get_asgi_application()
所以使用 AsgiHandler()
。但在未来可能会被废弃。
更多内容,参阅官方文档:https://channels.readthedocs.io/en/latest/releases/3.0.0.html
Demo
之前2.0的版本我写了一个类似聊天组的例子,虽然可以实现,但通知用户的话,其实一个组就可以,连接的用户都加入到这个组当中,然后不同的消息只让需要接收的人接收即可。Channels基础知识可以参照之前写的帖子。
下面是基于Channels3.0
实现的代码。
- my_project/settings.py(channels的设置)
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
'corsheaders',
# 添加channels应用
'channels',
]
# 频道层的缓存
CHANNEL_LAYERS = {
"default": {
"BACKEND": "channels_redis.core.RedisChannelLayer",
"CONFIG": {
# See https://aioredis.readthedocs.io/en/v1.3.1/api_reference.html
# "hosts": ["redis://:password@localhost:6379/3"], # Redis URI
"hosts": ["unix:///tmp/redis.sock?db=3&password=password"] # UNIX domain socket
},
},
}
通过UNIX domain socket
连接redis
会比TCP
的连接要快,需要在redis
当中设置一下。
修改/etc/redis.conf
。将unixsocket
和unixsocketperm
前的#
去掉,并将 unixsocketperm
的值由 700
改为 777
,否则将不能清理缓存 。
- my_project/asgi.py ( 建立asgi应用,并指定其要使用的路由 )
"""
ASGI config for otdr project.
It exposes the ASGI callable as a module-level variable named ``application``.
For more information on this file, see
https://docs.djangoproject.com/en/3.1/howto/deployment/asgi/
"""
import os
import otdr.routing as routing
from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import AllowedHostsOriginValidator
from django.core.asgi import get_asgi_application
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'otdr.settings')
application = ProtocolTypeRouter({
"http": get_asgi_application(),
"websocket": AllowedHostsOriginValidator(AuthMiddlewareStack(
URLRouter(
routing.websocket_urlpatterns
)))
})
- my_project/routing.py(路由的逻辑)
from django.urls import re_path
import utils.consumers as consumers
# ws://ip:path
websocket_urlpatterns = [
re_path(r'^ws/notifications$',consumers.NotificationsConsumer.as_asgi()),
]
- my_project/utils/consumers.py( 消费者的实现 )
import json
from channels.generic.websocket import AsyncWebsocketConsumer
class NotificationsConsumer(AsyncWebsocketConsumer):
"""异步处理通知的WebSocket请求"""
async def connect(self):
"""建立连接"""
await self.channel_layer.group_add('notifications', self.channel_name)
await self.accept()
async def receive(self, text_data=None, bytes_data=None):
"""将接收到的消息处理后返回给前端"""
if isinstance(text_data, dict):
text_data.pop("type","")
await self.send(text_data=json.dumps(text_data,ensure_ascii=False))
async def disconnect(self, code):
"""断开连接"""
await self.channel_layer.group_discard('notifications', self.channel_name)
- my_project/utils/tools.py(生产者的实现 )
def notification_handler(json_data):
'''
通知处理器
:param json_data: json数据
:return: None
'''
json_data["type"] = "receive"
payload = json_data
channel_layer = get_channel_layer()
async_to_sync(channel_layer.group_send)('notifications', payload)
后面只需要在视图层调用notification_handler()
这个函数,服务端即可主动推送通知给前端了!
每次看到你的文章,我都觉得时间过得好快。 https://www.4006400989.com/qyvideo/64640.html
《归徒》剧情片高清在线免费观看:https://www.jgz518.com/xingkong/142551.html