将channels加入到INSTALLED_APPS配置中：
INSTALLED_APPS = (
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.sites',
'channels',
创建一个默认路由文件，与Django的路由文件区别，为myproject/routing.py:
from channels.routing import ProtocolTypeRouter
application = ProtocolTypeRouter({
最后，在配置文件(settings.py)通过ASGI_APPLICATION配置将第二步中设置的ASGI_APPLICATION作为我们的根应用：
ASGI_APPLICATION = "myproject.routing.application"
一旦配置完毕，channel将会被集成在Django中，并且会接管runserve命令。
假设我们现已有一个部分功能开发完成的Django项目，现在要求服务端能够对不同用户打开的web客户端发送实时消息。即建立的websocket链接能够按照已经验证过的user分组，并且能在我们已有的视图文件(views.py)中向建立的websocket发送消息。
接下来开始coding。
按照模块化的思想，首先创建一个应用来完成我们的功能。这一步和正常的Django创建应用相同
startapp:
python manage.py startapp websockets
将应用添加到INSTALLED_APPS：
# myproject/settings.py
INSTALLED_APPS = [
'django.contrib.admin',
'django.contrib.auth',
'django.contrib.contenttypes',
'django.contrib.sessions',
'django.contrib.messages',
'django.contrib.staticfiles',
“channels",
‘'websockets'
routing
与Django的urls.py文件功能类似，我们同样需要将创建的websocket链接绑定到不同的处理函数上。
ProtocolTypeRouter
因为channels支持HTTP、WebSockets等协议，所以我们需要在根应用路由中拥有一个ProtocolTypeRouter来将不同的协议通过协议服务器转到不同的子路由中处理：
from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
import websockets.routing
application = ProtocolTypeRouter({
'websocket': AuthMiddlewareStack(
URLRouter(
websockets.routing.websocket_urlpatterns
URLRouter
URLRouter通过请求路径路由不同的http或websocket链接，是一个Dajngo url列表。
接上一步，需要在websockets应用中创建routing.py文件然后根据路径定义url列表，即websocket_urlpatterns：
from django.conf.urls import url
from . import consumers
websocket_urlpatterns = [
url(r'^ws/$', consumers.WsConsumer),
ChannelNameRouter
还有一类路由通过channel key在scope中的值来进行路由，在本例中用不上，可参见官方文档：
ChannelNameRouter({
"thumbnails-generate": some_app,
"thunbnails-delete": some_other_app,
到这一步路由就完成了，接下来就是定义我们的处理类，也就是WsConsumer.
Consumers
在websockets目录下新建consumers.py文件：
from asgiref.sync import async_to_sync
from channels.generic.websocket import WebsocketConsumer
import json
class WsConsumer(WebsocketConsumer):
def connect(self):
self.user = self.scope["user"]
if self.user.is_anonymous:
self.close()
return
self.group_name = 'task_%s' % str(self.user.id)
async_to_sync(self.channel_layer.group_add)(
self.group_name,
self.channel_name
self.accept()
def disconnect(self, close_code):
if not self.user.is_anonymous:
async_to_sync(self.channel_layer.group_discard)(
self.group_name,
self.channel_name
def receive(self, text_data):
text_data_json = json.loads(text_data)
message = text_data_json['message']
async_to_sync(self.channel_layer.group_send)(
self.group_name,
'type': 'task_message',
'message': message
def task_message(self, event):
message = event['message']
self.send(text_data=json.dumps({
'message': message
在Consumers中，我们构造了一系列当事件发生时就会对应调用的函数, 而不是去编写事件循环来进行事件处理。
Generic Consumers
在代码中，相比于继承channels.consumer.AsyncConsumer或channels.consumer.SyncConsumer来创建基本的consumer，我们继承了包装更为完善的Generic Consumers：WebsocketConsumer。
其他的Generic Consumers还有AsyncWebsocketConsumer、JsonWebsocketConsumer、 AsyncJsonWebsocketConsumer和AsyncHttpConsumer，正如字面意义，都有同步异步之分，这里不一一介绍，可详见官方文档。
connect
当前端发送一个新建websocket的请求过来的时候，会自动触发connect()函数事件。
<script>
var webSocket = new WebSocket(
'ws://' + window.location.host + '/ws/');
</script>
disconnet
当 server 端的 WebSocket 调用 self.close() 关闭时，前端的 webSocket.onclose 会自动触发
<script>
webSocket.onclose = function(e) {
console.error('web socket closed unexpectedly');
</script>
而当链接断开时，无论是主动的前端页面关闭，还是被动的网络故障，都会触发disconnect()函数事件。
当然若因断电或其他故障导致的服务挂掉，自然也就触发不了disconnect()函数。
receive
当前端或后端将信息发送到websocket中后，服务端的receive()事件就会被触发，可在此函数中对收到的消息进行处理。在上面我们定义的文件中，我们将收到的消息通过group(后续介绍)群发到group包含的通道中，然后每个通道通过task_message函数对消息进行处理。
前端发送请求：
<script>
document.querySelector('#task-message-submit').onclick = function(e) {
var messageInputDom = document.querySelector('#task-message-input');
var message = messageInputDom.value;
webSocket.send(JSON.stringify({
'message': message
messageInputDom.value = '';
</script>
task_message
自定义消息处理函数。当通道收到消息调用receive()函数时，若我们只需要此通道对消息进行处理，自然也可以在receive()函数中直接进行处理。但当我们需要通道所属group的其他成员也能对消息进行处理，那么只在receive()函数中处理就满足不了需要了。
通过将消息群发到每个通道的task_message中进行处理。
Scope
consumer 在初始化连接时收到连接的scope, 其中包含大部分我们能在Django视图中的Request对象上找到的信息。在consumer的方法内通过self.scope进行调用。
scope是ASGI规范的一部分, 但以下是我们可能要使用的一些常见内容:
scope["path"], the path on the request. (HTTP and WebSocket)
scope["headers"], raw name/value header pairs from the request (HTTP and WebSocket)
scope["method"], the method name used for the request. (HTTP)
在我们的例子中，因为在路由中使用的是AuthMiddlewareStack来处理websocket请求，所以我们可以直接通过scope["user"]取出当前登录的user信息
Channel Layers
试想想，在项目中，我们不仅需要在收到前端消息时能对消息进行处理，而且在Django的其他业务处理视图中也需要向前端发现消息。比如在Django业务处理逻辑中，新增了某个用户或某类用户的待处理任务，我们怎么在Django的业务逻辑中（脱离了consumer）给通道内广播信息呢？
通道层允许我们在应用程序的不同实例之间进行对话。如果您不想将所有消息或事件都传送到数据库中, 它们是制作分布式实时应用程序的一个有用部分。
channel layers 因为是使用 redis 作为消息通道层，这使得一个应用中两个实例之间可以实现通信。使用channel layers需要redis的支持。首先需要在配置文件配置所需要连接的redis服务。
CHANNEL_LAYERS = {
"default": {
"BACKEND": "channels_redis.core.RedisChannelLayer",
"CONFIG": {
"hosts": [("redis-server-name", 6379)],
在consumer内部使用self.channel_layer来获取channel layers的实例。
在脱离了consumer的情况下，Channerls提供了get_channel_layer函数接口来获取它。
from channels.layers import get_channel_layer
channel_layer = get_channel_layer()
一旦你获取到了实例，你就可以调用实例中的方法。但请记住，channel layers仅支持异步方法，因此你可以从自己的异步上下文中调用它：
for group in groups:
await channel_layer.group_send(
group_name,
{"type": "task.message", "message": "Hello there!"},
或者，如果你的上下文是同步环境，可以使用 async_to_sync 来转换调用：
from asgiref.sync import async_to_sync
async_to_sync(channel_layer.group_send)(
group_name,
{"type": "task.message", "message": "Hello there!"}
Single Channels
channel_layer.send
当我们不需要群发，而是向某个通道发送消息时。只需要找到通道的名字，使用channel_layer.send:
from channels.layers import get_channel_layer
channel_layer = get_channel_layer()
await channel_layer.send("channel_name", {
"type": "task.message",
"message": "Hello there!",
type标识处理消息的函数，也就是这里的消息会被consumer中的task_message函数处理。type后的键值对会形成一个字典作为参数传给type所标识的函数
注意：这里使用"type": "task.message"与"type": "task_message"效果一致，channels会将.转换为_。
channel_name
channel_name是每一个websocket请求过来时候， Channels 自动帮我们生成的一个个性化名称，实现代码就是用随机函数组装：
async def new_channel(self, prefix="specific."):
Returns a new channel name that can be used by something in our
process as a specific channel.
return "%s.inmemory!%s" % (
prefix,
"".join(random.choice(string.ascii_letters) for i in range(12)),
Group
显然, 发送到单个频道并不是特别有用--在大多数情况下, 我们希望作为广播同时发送给多个频道consumer。
channel_layer.group_add
将频道加入到某个group中，如我们的例子中：
async_to_sync(self.channel_layer.group_add)(
self.group_name,
self.channel_name
第一个参数为group的名字，第二个参数为channel_name.
channel_layer.group_discard
将频道从group中移除，如我们的例子中：
async_to_sync(self.channel_layer.group_discard)(
self.group_name,
self.channel_name
channel_layer.group_send
向group中的频道群发消息：
async_to_sync(self.channel_layer.group_send)(
self.group_name,
'type': 'task_message',
'message': message
至此，我们的consumer.py里所牵涉的内容基本都介绍完毕了。
Authentication
Channels中的AuthMiddleware支持标准的 Django身份验证, 其中用户详细信息存储在session中。它允许对scope中的用户对象进行只读访问。
若要使用中间件, 请将其包装在路由中适当级别的consumer周围:
from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
import websockets.routing
application = ProtocolTypeRouter({
'websocket': AuthMiddlewareStack(
URLRouter(
websockets.routing.websocket_urlpatterns
使用self.scope["user"]来获取登录的用户：
class WsConsumer(WebsocketConsumer):
def connect(self):
self.user = self.scope["user"]
注意，因为用户详细信息存储在session中。若我们使用了token的认证方式，前端通过token来验证用户，而不是用session，那么我们需要自定义我们的认证文件。
以jwt认证方式来举例，新建ws_authentication.py文件：
from functools import wraps
import jwt
import json
import logging
from django.utils.translation import ugettext_lazy as _
from django.contrib.auth import get_user_model
from rest_framework import exceptions
from rest_framework_jwt.settings import api_settings
import logging
logger = logging.getLogger(__name__)
User = get_user_model()
jwt_payload_handler = api_settings.JWT_PAYLOAD_HANDLER
jwt_encode_handler = api_settings.JWT_ENCODE_HANDLER
jwt_decode_handler = api_settings.JWT_DECODE_HANDLER
jwt_get_username_from_payload = api_settings.JWT_PAYLOAD_GET_USERNAME_HANDLER
def token_authenticate(token, message):
Tries to authenticate user based on the supplied token. It also checks
the token structure and validity.
payload = check_payload(token=token, message=message)
user = check_user(payload=payload, message=message)
"""Other authenticate operation"""
return user, token
def check_payload(token, message):
payload = None
try:
payload = jwt_decode_handler(token)
except jwt.ExpiredSignature:
msg = _('Signature has expired.')
logger.warn(msg)
except jwt.DecodeError:
msg = _('Error decoding signature.')
logger.warn(msg)
return payload
def check_user(payload, message):
username = None
user = None
try:
username = payload.get('username')
except Exception:
msg = _('Invalid payload.')
logger.warn(msg)
_close_reply_channel(message)
if not username:
msg = _('Invalid payload.')
logger.warn(msg)
_close_reply_channel(message)
return
try:
user = User.objects.get_by_natural_key(username)
except User.DoesNotExist:
msg = _("User doesn't exist.")
logger.warn(msg)
return user
raise exceptions.AuthenticationFailed(msg)
if not user.is_active:
msg = _('User account is disabled.')
logger.warn(msg)
raise exceptions.AuthenticationFailed(msg)
return user
def _close_reply_channel(message):
return
def ws_auth_request_token(func):
Checks the presence of a "token" request parameter and tries to
authenticate the user based on its content.
The request url must include token.
eg: /v1/channel/1/?token=abcdefghijklmn
@wraps(func)
def inner(message, *args, **kwargs):
scope = message.scope
token = scope.get("query_string").decode().split("=", 1)[1]
if token is None:
_close_reply_channel(message)
raise ValueError("Missing token request parameter. Closing channel.")
user, token = token_authenticate(token, message)
message.token = token
message.user = user
return func(message, *args, **kwargs)
return inner
然后在我们需要进行验证的函数前，使用该装饰器:
class WsConsumer(WebsocketConsumer):
@ws_auth_request_token
def connect(self):
该装饰器会将前端对前端新建链接url中的token参数进行jwt验证，将验证通过的user赋值给scope['user']
到这，基本就完成了我们需求中的要求。更多细节就不再介绍。
有关更多channel细节，可参见官方文档
2024-08-01 14:07:45
