Channels集成到Django消息實時推送

channel架構圖

InterFace Server：負責對協議進行解析，將不同的協議分發到不同的Channel

Channel Layer：頻道層，可以是一個FIFO隊列，通常使用Redis

 

Django中配置Channel：

CHANNEL_LAYERS的配置：

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": ["redis://127.0.0.1:6379", ], 
        },
    },
}

asgi的配置：

import os
import sys
import django
from channels.routing import get_default_application


os.environ.setdefault("DJANGO_SETTINGS_MODULE", "web.settings_local")
django.setup()
application = get_default_application()

consumers代碼：

class NotificationsConsumer(AsyncWebsocketConsumer):
    """處理通知應用中的WebSocket請求"""

    async def connect(self):
        """建立連接"""
        if self.scope['user'].is_anonymous:
            # 未登錄用戶拒絕連接
            await self.close()
        else:
            await self.channel_layer.group_add(
                'notifications', self.channel_name
            )
            await self.accept()

    async def receive(self, text_data=None, bytes_data=None):
        """將接收到的消息返回給前端"""
        await self.send(text_data=json.dumps(text_data))

    async def disconnect(self, code):
        """斷開連接"""
        await self.channel_layer.group_discard(
            'notifications', self.channel_name
        )

 channels將同步的MySQL轉換為異步的：

# ORM語句同步變異步，方式一
from channels.db import database_sync_to_async
user = await database_sync_to_async(User.objects.get(username=username))

# ORM語句同步變異步，方式二
@database_sync_to_async
def get_username(username):
    return User.objects.get(username=username)

配置routing：

from django.urls import path
from channels.auth import AuthMiddlewareStack  # channels的認證中間件
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import AllowedHostsOriginValidator

from notifications.consumers import NotificationsConsumer


application = ProtocolTypeRouter({
    'websocket': AllowedHostsOriginValidator(
        AuthMiddlewareStack(
            URLRouter([
                path('ws/notifications/', NotificationsConsumer),
            ])
        )
    )
})

settings中配置：

INSTALLED_APPS中加入：

'channels',

配置ASGI_APPLICATION：
ASGI_APPLICATION = 'web.routing.application'

 

notifications 業務層實現邏輯：

channel_layer = get_channel_layer()
payload = {
    'type': 'receive',
    'key': key,
    'actor_name': actor.username,
    'id_value': id_value
}
async_to_sync(channel_layer.group_send)('notifications', payload)

消息通知業務流程：
用戶觸發了消息 --> Django的view層 --> 保存到MySQL數據庫 --> 將消息通知發送到channel對應的group里面 --> websocket將消息通過consumer推送給接收方