Django websocket 进行实时通信（消费者）

1. settings.py 增加

ASGI_APPLICATION = "django_template_v1.routing.application"

CHANNEL_LAYERS = {
    "default": {
        # This example apps uses the Redis channel layer implementation channels_redis
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": ["{}0".format(REDIS_URL)],
        },
    },
}

2. 跟settings.py同级目录下，添加routing.py文件

from django.urls import path, re_path
from channels.auth import AuthMiddlewareStack
from channels.routing import ProtocolTypeRouter, URLRouter
from apps.message_manage.consumers import ChatConsumer, NotificationChatConsumer

# QueryAuthMiddlewareStack、AuthMiddlewareStack

application = ProtocolTypeRouter({
    "websocket": URLRouter(
        [
            re_path(r'^ws/chat/(?P<recipient>\w+)/$', ChatConsumer),
            # re_path(r'^ws/chatting/(?P<recipient>\w+)/$', NotificationChatConsumer),
            re_path(r'^ws/chatting/(?P<recipient>\w+)/(?P<platform_key>\w+)/$', NotificationChatConsumer),
        ]
    ),
})

3. 跟settings.py同级目录下，添加asgi.py文件

"""
ASGI config for django_template_v1 project.

It exposes the ASGI callable as a module-level variable named ``application``.

For more information on this file, see
https://docs.djangoproject.com/en/3.2/howto/deployment/asgi/
"""

import os
import django
from channels.routing import get_default_application

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'django_template_v1.settings')
django.setup()
application = get_default_application()

4. 编写消费者，添加consumers.py文件（不是和settings.py一个目录了）

# -*- coding: utf-8 -*-
"""
-------------------------------------------------
   File Name：      consumers
   Description:
   Author:          Administrator
   date：           2018/6/6
-------------------------------------------------
   Change Activity:
                    2018/6/6:
-------------------------------------------------
"""
import json
import logging
import aioredis
import asyncio
from channels.generic.websocket import AsyncWebsocketConsumer
from django_template_v1.settings import REDIS_URL

logger = logging.getLogger("websocket")


class ChatConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.recipient = self.scope['url_route']['kwargs']['recipient']
        logger.info(f"websocket建立连接成功，连接的用户={self.recipient}")

        self.room_group_name = f'{self.recipient}'

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

        await self.accept()

        # 执行定时任务
        self.periodic_task = asyncio.create_task(self.send_periodic_message())

        # 获取并发送历史未读消息
        # unread_notifications = NotificationRecord.objects.filter(recipient=self.recipient, is_read=False)
        # for notification in unread_notifications:
        #     await self.send(text_data=json.dumps({
        #         'message': notification.subject,
        #         'is_unread': True,
        #         'recipient': self.recipient,
        #         "receive_time": notification.receive_time.strftime('%Y-%m-%d %H:%M:%S')
        #     }))
        #     notification.is_read = True
        #     notification.save()

    async def disconnect(self, close_code):
        # Leave room group
        print(f"disconnect websocket")
        if hasattr(self, 'periodic_task') and not self.periodic_task.done():
            self.periodic_task.cancel()

        await self.channel_layer.group_discard(
            self.room_group_name,
            self.channel_name
        )

        await super().disconnect(close_code)

    # Receive message from WebSocket
    async def receive(self, text_data):
        text_data_json = json.loads(text_data)
        message = text_data_json['message']
        print(f"Received message: {message}")

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

    async def send_periodic_message(self):
        """Periodically sends a message to the client."""
        while True:
            await asyncio.sleep(10)  # Sleep for a minute
            await self.send(text_data=json.dumps({
                'message': f'每隔一分钟的消息: {self.recipient}'
            }))

    async def chat_message(self, event):
        """Handler for messages sent through channel layer."""
        message = '测试聊天室：' + event['message']

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


class NotificationChatConsumer(AsyncWebsocketConsumer):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.redis_conn = None
        self.online_users_set_name = None
        self.room_group_name = None

    async def connect(self):
        self.recipient = self.scope['url_route']['kwargs']['recipient']
        self.platform_key = self.scope['url_route']['kwargs']['platform_key']
        logger.info(f"WebSocket建立连接成功, recipient: {self.recipient}, platform_key: {self.platform_key}")

        # 使用aioredis创建异步连接
        self.redis_conn = await aioredis.from_url("{}12".format(REDIS_URL))
        logger.info(f"使用aioredis 进行redis连接")

        # 构建特定于平台的在线用户集合名称
        self.online_users_set_name = f'online_users_{self.platform_key}'

        # 异步添加recipient到特定于平台的online_users集合
        await self.redis_conn.sadd(self.online_users_set_name, self.recipient)
        logger.info(f"websocket 添加recipient到{self.online_users_set_name}集合")

        self.room_group_name = f'{self.recipient}_{self.platform_key}'

        # 加入room组
        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name
        )

        await self.accept()

    async def disconnect(self, close_code):
        logger.info(f"disconnect WebSocket, close_code: {close_code}")
        if self.redis_conn and self.recipient:
            logger.info(f"websocket disconnect，从{self.online_users_set_name}集合移除{self.recipient}")
            await self.redis_conn.srem(self.online_users_set_name, self.recipient)

        # 离开room组
        await self.channel_layer.group_discard(
            self.room_group_name,
            self.channel_name
        )

        # 关闭Redis连接
        if self.redis_conn:
            self.redis_conn.close()
            await self.redis_conn.wait_closed()

    # Receive message from WebSocket
    async def receive(self, text_data):
        text_data_json = json.loads(text_data)
        message = text_data_json['message']
        logger.info(f"Received message: {message}")

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

    async def notification_message(self, event):
        logger.info(f"notification_message: {event}")
        # 直接发送event作为消息内容
        await self.send(text_data=json.dumps(event))

5. 进行实时发送，添加task.py文件

from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync

channel_layer = get_channel_layer()

# 发送websocket消息
def send_ws_msg(group_name, data):
    async_to_sync(channel_layer.group_send)(
        group_name,
        {'type': 'notification_message', 'message': data}
    )
    return



def test_send(account, platform_key, message_dict):
	send_ws_msg(f"{account}_{platform_key}", message_dict)