ASGI
认识 Channels 之前，需要先了解一下 asgi ，全名：Asynchronous Server Gateway Interface。它是区别于 wsgi 的一种异步服务网关接口，不仅仅只是通过 asyncio 以异步方式运行，还支持多种协议。完整的文档戳这里
关联的几个项目：
- https://github.com/django/asgiref ASGI内存中的通道层，函数的同步异步之间相互转化需要
- https://github.com/django/daphne 支持HTTP，HTTP2和WebSocket协议服务器，启动 Channels 的项目需要
- https://github.com/django/channels_redis Channels专属的通道层，使用Redis作为其后备存储，并支持单服务器和分片配置以及群组支持。（这个项目是 Channels 的一个附属项目，配置的时候作为可选项使用，该软件包的早期版本被称为 asgi_redis，如果你在使用 Channels 1.x项目，它仍可在PyPI下通过这个名称使用。但 channels_redis 仅适用于 Channels 2 项目。）
备注：之前体验使用过一个基于 asgi 的web 框架 uvicon，也是同类项目。
流程图
流程图
后端实现
- 安装channles
pip install -U channels
pip install -U channels_redis
pip install -U django-redis
- APP上加上channels
# settings.py
INSTALLED_APPS = [
'channels', # 加上channels
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
]
- 添加asgi路由
# settings.py
ASGI_APPLICATION = "myWebSocket.routing.application"
- 添加channles对应的数据库引擎
# settings.py
# 【channels】后端
# redis标准配置 ["redis://password/host/db"]
CHANNEL_LAYERS = {
"default": {
"BACKEND": "channels_redis.core.RedisChannelLayer",
"CONFIG": {
"hosts": ["redis://192.168.0.112:6379/0"], ## redis地址
},
},
}
- consumer后端实现
## consumer
# 【channels】（第4步）创建应用的消费者
from channels.generic.websocket import WebsocketConsumer, AsyncWebsocketConsumer
from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
import json
# 使用异步时继承自AsyncWebsocketConsumer而不是WebsocketConsumer。
# 所有方法都是async def而不是def。
# await用于调用执行I / O的异步函数。
# 在通道层上调用方法时不再需要async_to_sync。
class AsyncConsumer(AsyncWebsocketConsumer):
async def connect(self): # 连接时触发
self.room_name = self.scope['url_route']['kwargs']['room_name']
self.room_group_name = 'notice_%s' % self.room_name # 直接从用户指定的房间名称构造Channels组名称，不进行任何引用或转义。
# 将新的连接加入到群组
await self.channel_layer.group_add(
self.room_group_name,
self.channel_name
)
await self.accept()
async def disconnect(self, close_code): # 断开时触发
# 将关闭的连接从群组中移除
await self.channel_layer.group_discard(
self.room_group_name,
self.channel_name
)
# Receive message from WebSocket
async def receive(self, text_data=None, bytes_data=None): # 接收消息时触发
text_data_json = json.loads(text_data)
message = text_data_json['message']
# 信息群发
await self.channel_layer.group_send(
self.room_group_name,
{
'type': 'system_message',
'message': message
}
)
# Receive message from room group
async def system_message(self, event):
print(event)
message = event['message']
# Send message to WebSocket单发消息
await self.send(text_data=json.dumps({
'message': message
}))
# 同步方式，仅作示例，不使用
class SyncConsumer(WebsocketConsumer):
def connect(self):
# 从打开到使用者的WebSocket连接的chat/routing.py中的URL路由中获取'room_name'参数。
self.room_name = self.scope['url_route']['kwargs']['room_name']
print('WebSocket建立连接：', self.room_name)
# 直接从用户指定的房间名称构造通道组名称
self.room_group_name = 'msg_%s' % self.room_name
# 加入房间
async_to_sync(self.channel_layer.group_add)(
self.room_group_name,
self.channel_name
) # async_to_sync(…)包装器是必需的，因为ChatConsumer是同步WebsocketConsumer，但它调用的是异步通道层方法。(所有通道层方法都是异步的。)
# 接受WebSocket连接。
self.accept()
simple_username = self.scope["session"]["session_simple_nick_name"] # 获取session中的值
async_to_sync(self.channel_layer.group_send)(
self.room_group_name,
{
'type': 'chat_message',
'message': '@{} 已加入房间'.format(simple_username)
}
)
def disconnect(self, close_code):
print('WebSocket关闭连接')
# 离开房间
async_to_sync(self.channel_layer.group_discard)(
self.room_group_name,
self.channel_name
)
# 从WebSocket中接收消息
def receive(self, text_data=None, bytes_data=None):
print('WebSocket接收消息：', text_data)
text_data_json = json.loads(text_data)
message = text_data_json['message']
# 发送消息到房间
async_to_sync(self.channel_layer.group_send)(
self.room_group_name,
{
'type': 'chat_message',
'message': message
}
)
# 从房间中接收消息
def chat_message(self, event):
message = event['message']
# 发送消息到WebSocket
self.send(text_data=json.dumps({
'message': message
}))
- asgi主路由配置
# 工程主项目下新增routing.py, 与setttings.py上ASGI_APPLICATION变量值对应即可
# 【channels】（第5步）设置默认路由在项目创建routing.py文件
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from channels.sessions import SessionMiddlewareStack
import ws1.routing
application = ProtocolTypeRouter({
# (http->django views is added by default)
# 【channels】（第6步）添加路由配置指向应用的路由模块
'websocket': SessionMiddlewareStack( # 使用Session中间件，可以请求中session的值
URLRouter(
ws1.routing.websocket_urlpatterns
)
),
})
- 为应用程序创建一个路由配置
# ws1/routing.py，与上个步骤定下的路由值对应即可
# 【channels】（第6步）为应用程序创建一个路由配置，该应用程序具有到消费者的路由
from django.conf.urls import url
from . import consumer
websocket_urlpatterns = [
# url(r'^ws/msg/(?P<room_name>[^/]+)/$', consumers.SyncConsumer),
url(r'^ws/msg/(?P<room_name>[^/]+)/$', consumer.AsyncConsumer),
]
- ASGI 接口
# 工程主项目下新增asgi.py, 与wsgi.py类似
# 项目/settings和wsgi.py的同目录下创建asgi.py
"""
ASGI入口点，运行Django，然后运行在settings.py ASGI_APPLICATION 中定义的应用程序
安装：pip install daphne
运行：daphne -p 8001 ITNest.asgi:application
"""
import os
import django
from channels.routing import get_default_application
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myWebSocket.settings")
django.setup()
application = get_default_application()
- 安装daphne，启动channels
pip install -U daphne
daphne -p 8000 myWebSocket.asgi:application
前端实现
<head>
<meta http-equiv="Content-Type" content="text/html;charset=UTF-8">
</head>
<link href="css/toastr.min.css" rel="stylesheet">
<script src="js/jquery-1.10.2.min.js"></script>
<script src="js/toastr.min.js"></script>
<script>
let ws_scheme = window.location.protocol === "https:" ? "wss" : "ws";
let ws_port = 8000;
let ws = new WebSocket(ws_scheme + '://' + window.location.host + ':' + ws_port + '/ws/msg/ITNest/');
ws.onopen = function () {
console.log('WebSocket建立连接');
};
ws.onmessage = function (e) {
console.log('WebSocket接收消息：');
let data = JSON.parse(e.data);
console.log(data);
let message = data['message'];
if (message.level === 1) {
toastr.options = { // toastr配置
"closeButton": true,
"debug": false,
"progressBar": true,
"positionClass": "toast-top-center",
"showDuration": "400",
"hideDuration": "1000",
"timeOut": "30000",
"extendedTimeOut": "1000",
"showEasing": "swing",
"hideEasing": "linear",
"showMethod": "fadeIn",
"hideMethod": "fadeOut"
};
toastr.error(message.content, '警告');
} else {
toastr.options = { // toastr配置
"closeButton": true,
"debug": false,
"progressBar": true,
"positionClass": "toast-top-right",
"showDuration": "400",
"hideDuration": "1000",
"timeOut": "7000",
"extendedTimeOut": "1000",
"showEasing": "swing",
"hideEasing": "linear",
"showMethod": "fadeIn",
"hideMethod": "fadeOut"
};
toastr.info(message.content, '信息')
}
};
ws.onclose = function (e) {
console.error('WebSocket关闭连接');
};
</script>
django 调用websocket与前端方式消息
# 主程序代码段
def send_msg(room_name, message):
# 从Channels的外部发送消息给Channel
"""
from assets import consumers
consumers.send_group_msg('ITNest', {'content': '这台机器硬盘故障了', 'level': 1})
consumers.send_group_msg('ITNest', {'content': '正在安装系统', 'level': 2})
:param room_name:
:param message:
:return:
"""
channel_layer = get_channel_layer()
async_to_sync(channel_layer.send)(
'notice_{}'.format(room_name), # 构造Channels组名称
{
"type": "system_messages", # 不起作用
"message": message,
}
)
