Django Channels

优质

小牛编辑

135浏览

2023-12-01

Django Channels 是一个为 Django 提供异步扩展的库，通常主要用来提供 WebSocket 支持和后台任务。

原理

它的原理是将 Django 分为 2 种进程类型：

一个用于处理 HTTP 和 WebSocket 的协议服务
一个用于运行视图、WebSocket handler 以及后台任务的 worker 服务

两者通过 ASGI 协议通信，类似与 WSGI 但是运行在网络层上，并且支持更多协议。

Django Channels 并没有引入 asyncid、gevent 或者其它异步库，所有业务逻辑都在同一个业务进程/线程中同步运行。

细节请阅读 Django Channels docs/concepts

前提

想要运行 Django Channels 首先需要一个 ASGI 服务器，比如 Daphne，使用 Django worker 服务器运行：./manage.py runworker，以及 ASGI 请求的通道服务，比如 Redis。

即使启用 channels，所有的 HTTP 请求依旧是默认经过 Django 视图系统，因此可以无缝迁移过来。

好处

易用的支持数以千计客户端的 HTTP long-poll 支持
类似 Django 原生的 WebSocket 的会话和验证支持
自动根据 cookie 为 WebSocket 提供登录
大量事件触发底层支持
0 宕机部署支持
特定 URL 的底层 HTTP 控制
可扩展支持其它协议或者事件源（例如 WebRTC、UDP、SMS）
Django 的官方支持

基本概念

`channel`

Django Channels 的核心数据解构叫做 channel，它是一个有序的 FIFO 队列，支持信息过期，对于一个监听器至多投递一次。类似于任务队列──信息由 producer 投递至 channel，并设置一个 consumer 监听该 channel。

注意消息投递是至多一次，因此遇到应用崩溃等情况会导致消息丢失。

Django Channels 的理念和 Go 语言内置的 channel 概念类似，不同之处在于 Django Channels 是网络透明的，因此 producer 和 consumer 通信可以穿越不同机器。在同一个网络中，Django Channels 的频道是根据名称属性区分的，如果你向一个名为 http.request 的 channel 发送信息，都是发送向同一个 channel。而对于不同网络中则不一样，channel 是网络隔离的。

consumer

“consumer”（消费者）是 channels 架构中处理请求的部分，和 Django 有 view 函数和 ViewClass 一样，consumer 也支持类式写法。例如你要处理 WebSocket 请求，可以实现一个 WebsocketConsumer 的子类，并通过 route_class 函数将该 consumer 与 URL 连接起来。

consumer 函数

consumer 类

from channels.generic.websockets import WebsocketConsumer

class MyConsumer(WebsocketConsumer):
    http_user = True         # 设置为 ``True`` 将会自动从 HTTP cookie 中登录用户，因此可以省去 channel_session_user 的设置。
    strict_ordering = False  # 默认设置

    def connection_groups(self, **kw):
        """返回 group 列表，并自动将本连接插入其中活删除
        """
        return ['test']

    def connect(self, message, **kw):
        """连接开始
        """
        self.message.reply_channel.send({'accept': True})

    def receive(self, text=None, bytes=None, **kw):
        """接收到信息时调用的函数
        """
        self.send(text=text, bytes=bytes)  # 将信息原封不动地返回

    def disconnect(self, message, **kw):
        """断开连接时将会被调用
        """
        pass

`JsonWebsocketConsumer`

from channels.generic.websockets import JsonWebsocketConsumer

class MyConsumer(JsonWebsocketConsumer):
    """使用与 WebsocketConsumer 基本一致
    """
    def receive(self, content, **kwargs):
        """这里的 content 会是 JSON 解码后的 Python 对象
        """
        self.send(content)

    # JsonWebsocketConsumer 实际上在接收到信息后调用了下面的方法，你也可以重载它们来调用自己的 JSON 编、解码器
    # @classmethod
    # def decode_json(cls, text):
    #     return my_custom_json_decoder(text)
    #
    # @classmethod
    # def encode_json(cls, content):
    #     return my_custom_json_encoder(content)

多通道 WebSocket

channels 还支持多个数据复用同一个 WebSocket，只需要使用 Demultiplexer 来分流。

from channels.generic.websockets import WebsocketDemultiplexer, JsonWebsocketConsumer

class EchoConsumer(JsonWebsocketConsumer):
    def connect(self, message, multiplexer, **kwargs):
        # 通过 multiplexer 发送消息
        multiplexer.send({"status": "I just connected!"})

    def disconnect(self, message, multiplexer, **kwargs):
        print("Stream %s is closed" % multiplexer.stream)

    def receive(self, content, multiplexer, **kwargs):
        multiplexer.send({"original_message": content})  # 原封不动地返回消息

class AnotherConsumer(JsonWebsocketConsumer):
    def receive(self, content, multiplexer=None, **kwargs):
        # 你的实现
        pass

class Demultiplexer(WebsocketDemultiplexer):
    # Wire your JSON consumers here: {stream_name : consumer}
    consumers = {
        "echo": EchoConsumer,
        "other": AnotherConsumer,
    }

`Group`

会话和用户

如上面所提到的，你可以通过设置 consumer 类的 channel_session 或者 channel_session_user 属性来设置用户会话。也可以设置 http_user 使用 Django 的 Session 和 User。之后便可以获取 message 对象的 channel_session 和 user 属性了。

对于函数 consumer，则可以使用 channels.auth.channel_session_user 等装饰器达到相同效果：

from channels.routing import route
from channels import Group
from channels.auth import channel_session_user, channel_session_user_from_http

@channel_session_user_from_http
def connect(message):
    pass

@channel_session_user
def disconnect(message):
    pass

channel_routing = [
    route('websocket.connect', connect),
    route('websocket.disconnect', disconnect),
]

backend

backend 是为 Django Channels 提供 channel 异步通信的后端，推荐使用 Redis 作为 backend。

根据 backend 的不同，channels 的行为可能会有所不同，比如 asgi_ipc.IPCChannelLayer 作为 backend 部署起来更简单，但是并不支持网络间通信。asgi_redis.RedisChannelLayer 虽然支持网络间通信，但是需要额外的 Redis 服务器支持。

Django Channels 支持 Redis、IPC 以及内存间通信，开发者可以根据应用场景和部署环境灵活选择 backend。

部署

设置 backend

需要在 Django 的 settings 里定义 CHANNEL_LAYERS：

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "asgi_redis.RedisChannelLayer",      # 使用 asgi_redis 作为 backend
        "ROUTING": "my_project.routing.channel_routing",   # 指定路由文件
        "CONFIG": {
            "hosts": [("localhost", 6379)],             # 设置 Redis 的地址和端口
        },
    },
}

运行 worker 服务器

前面也提到了，Django Channels 将服务器解耦为协议服务器和 worker 服务器，因此也推荐单独部署 worker 服务器：python manage.py runworker。

如果有需要，你甚至可以将不同的 channel 分别部署，Django Channels 支持 --only-channels 和 --exclude-channels 用于过滤：

python manage.py runworker --only-channels=http.* --only-channels=websocket.*
python manage.py runworker --exclude-channels=thumbnail

运行接口服务器

由于 Django Channels 实现的是 ASGI 协议而不是 WSGI，因此你需要使用 ASGI 服务器来运行 channels 应用，例如 Daphne。使用 Daphne 还有一个好处就是，它会自动区分 HTTP 请求和 WebSocket 请求，因此并不需要分别启动两个服务器。

和 WSGI 应用部署类似，部署 ASGI 应用也需要首先指定 ASGI 应用本身，我们可以仿照 Django wsgi.py 的写法创建一个这样的 asgi.py 文件：

# coding: utf-8

import os
from channels.asgi import get_channel_layer

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "my_project.settings")

channel_layer = get_channel_layer()

然后运行 Daphne：daphne my_project.asgi:channel_layer