channels使用

安装django及channels

pip install channels

创建channels库根路由配置文件,根路由配置文件类似Django URLconf,它会告诉Channels当收到由Channes服务器发过来的Http请求时,应该执行什么代码:

# wssite/routing.py
from channels.routing import ProtocolTypeRouter

application = ProtocolTypeRouter({
    # (http->django views is added by default)
})

将 Channels 库添加到已安装的应用程序列表中。编辑 wssite/settings.py 文件并将 'channels' 添加到 INSTALLED_APPS 设置:

# mysite/settings.py
INSTALLED_APPS = [
    'channels',
    'chat',
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
]

配置根路由指向 Channels:

# wssite/settings.py
# Channels
ASGI_APPLICATION = 'wssite.routing.application'

现在已安装的应用程序中有 Channels, 它将控制 runserver 命令, 用 Channels 开发服务器替换标准的 Django 开发服务器。

consumer配置

在app中创建新文件consumer,并添加如下代码:

# app/consumers.py
from channels.generic.websocket import WebsocketConsumer
import json

class ChatConsumer(WebsocketConsumer):
    def connect(self):
        self.accept()

    def disconnect(self, close_code):
        pass

    def receive(self, text_data=None, bytes_data=None):
        text_data_json = json.loads(text_data)
        message = text_data_json['message']

        self.send(text_data=json.dumps({
            'message': message

这是一个同步 WebSocket consumer, 它接受所有连接, 接收来自其客户端的消息, 并将这些消息回送到同一客户端。现在, 它不向同一个房间的其他客户端广播消息。

Channels 还支持编写异步 consumers 以提高性能。但是, 任何异步 consumers 都必须小心, 避免直接执行阻塞操作。

我们需要为 app 创建一个路由配置, 它有一个通往 consumer 的路由。创建新文件 app/routing.py。并写入以下代码:

# app/routing.py
from django.conf.urls import url

from . import consumers

websocket_urlpatterns = [
    url(r'^ws/chat/(?P<room_name>[^/]+)/$', consumers.ChatConsumer),
]

下一步是将根路由指向 chat.routing 模块。在 mysite/routing.py 中, 导入 AuthMiddlewareStack、URLRouter 和 chat.routing ;并在 ProtocolTypeRouter 列表中插入一个 "websocket" 键, 格式如下:

# app/routing.py
from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
import chat.routing

application = ProtocolTypeRouter({
    # (http->django views is added by default)
    'websocket': AuthMiddlewareStack(
        URLRouter(
            chat.routing.websocket_urlpatterns
        )
    ),
})

这个根路由配置指定,当与 Channels 开发服务器建立连接的时候, ProtocolTypeRouter 将首先检查连接的类型。如果是 WebSocket 连接 (ws://或 wss://), 则连接会交给 AuthMiddlewareStack。

AuthMiddlewareStack 将使用对当前经过身份验证的用户的引用来填充连接的 scope, 类似于 Django 的 AuthenticationMiddleware 用当前经过身份验证的用户填充视图函数的请求对象。然后连接将被给到 URLRouter。

根据提供的 url 模式, URLRouter 将检查连接的 HTTP 路径, 以将其路由指定到到特定的 consumer。

channel layer

channel layer 是一种通信系统。它允许多个 consumer 实例互相交谈, 以及与 Django 的其他部分进行通信。

channel layer 提供以下抽象:

  channel 是可以发送消息的邮箱。每个 channel 都有一个名称。任何有名称的 channel 都可以向 channel 发送消息。

  group 是一组相关的 channels。group 具有名称。任何具有名字的 group 都可以按名称向 group 中添加/删除 channel, 也可以向 group 中的所有 channel 发送消息。无法列举特定 group 中的 channel。

每个 consumer 实例都有一个自动生成的唯一的 channel 名称, 因此可以通过 channel layer 进行通信。

使用redis作为channel layer的后备存储。

安装 channels_redis, 以便 Channels 知道如何调用 redis。运行以下命令:

pip install channels_redis

在使用 channel layer 之前, 必须对其进行配置。编辑 wssite/settings.py 文件并将 CHANNEL_LAYERS 设置添加到底部:

# wssite/settings.py
# Channels
ASGI_APPLICATION = 'mysite.routing.application'
CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.core.RedisChannelLayer',
        'CONFIG': {
            "hosts": [('127.0.0.1', 6379)],
        },
    },
}

现在我们有了一个 channel layer, 让我们在 ChatConsumer 中使用它。将以下代码放在 chat/consumers.py 中, 替换旧代码:

# app/consumers.py
from asgiref.sync import async_to_sync
from channels.generic.websocket import WebsocketConsumer
import json

class ChatConsumer(WebsocketConsumer):
    def connect(self):
        self.room_name = self.scope['url_route']['kwargs']['room_name']  # scope中包含有关其连接的信息且在app/routes.py中的url路由中获取'room_name'参数
        self.room_group_name = 'chat_%s' % self.room_name # 构建channels_group名称

        # Join room group
        async_to_sync(self.channel_layer.group_add)(
            self.room_group_name,
            self.channel_name
        )

        self.accept() # 用于接受websocket连接,不调用则表示拒绝接收连接

    def disconnect(self, close_code):
        # 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):
        text_data_json = json.loads(text_data)
        message = text_data_json['message']

        # Send message to room group
        async_to_sync(self.channel_layer.group_send)(
            self.room_group_name,
            {
                'type': 'chat_message',
                'message': message
            }
        )

    # Receive message from room group
    def chat_message(self, event):
        message = event['message']

        # Send message to WebSocket
        self.send(text_data=json.dumps({
            'message': message
        }))

异步模式

重写 ChatConsumer 使其变为异步的。在 app/consumers.py 中输入以下代码:

# app/consumers.py
from channels.generic.websocket import AsyncWebsocketConsumer
import json
class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.room_name = self.scope['url_route']['kwargs']['room_name']
        self.room_group_name = 'chat_%s' % self.room_name

        # Join room group
        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name
        )

        await self.accept()

    async def disconnect(self, close_code):
        # Leave room group
        await self.channel_layer.group_discard(
            self.room_group_name,
            self.channel_name
        )

    # Receive message from WebSocket
    async def receive(self, text_data):
        text_data_json = json.loads(text_data)
        message = text_data_json['message']

        # Send message to room group
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                'type': 'chat_message',
                'message': message
            }
        )

    # Receive message from room group
    async def chat_message(self, event):
        message = event['message']

        # Send message to WebSocket
        await self.send(text_data=json.dumps({
            'message': message
        }))

这些用于 ChatConsumer 的新代码与原始代码非常相似, 它们具有以下差异:

  • 现在 ChatConsumer 继承自 AsyncWebsocketConsumer 而不是 WebsocketConsumer。
  • 所有方法都是 async def, 而不仅仅是 def。
  • await 被用于调用执行 I/O 的异步函数。
  • 在 channel layer 上调用方法时, 不再需要 async_to_sync。

数据库访问

Django ORM是一段同步代码,因此如果想从异步代码访问它,需要进行特殊处理以确保其连接正确关闭。

如果你正在使用SyncConsumer或者基于它的任何东西 - 比如 JsonWebsocketConsumer- 你不需要做任何特别的事情,因为所有代码都已经在同步模式下运行,并且Channels将作为SyncConsumer代码的一部分为你做清理工作。

但是,如果要编写异步代码,则需要使用安全的同步上下文调用数据库方法database_sync_to_async

数据库连接

如果使用线程使用者(同步的),通道可能会比可能使用的通道打开更多的数据库连接 - 每个线程最多可以打开一个连接。

默认情况下,线程数设置为“CPU数* 5”,因此可以看到最多这个线程数。如果要更改它,请将ASGI_THREADS环境变量设置为希望允许的最大数量。

为了避免在连接中有太多线程空闲,可以改写代码以使用异步使用者,并且只在需要使用Django的ORM(使用database_sync_to_async)时才进入线程。

database_sync_to_async

要使用它,请在单独的函数或方法中编写ORM查询,然后database_sync_to_async像这样调用它:

from channels.db import database_sync_to_async

async def connect(self):
    self.username = await database_sync_to_async(self.get_name)()

def get_name(self):
    return User.objects.all()[0].name

还可以将它用作装饰器:

from channels.db import database_sync_to_async

async def connect(self):
    self.username = await self.get_name()

@database_sync_to_async
def get_name(self):
    return User.objects.all()[0].name
原文地址:https://www.cnblogs.com/Wunsch/p/12366298.html