如何获取WebSocket中async_to_sync调用的日志、结果及通道连接客户端数

一、确认channel_layer.group_send调用正常执行 #

1. 捕获调用异常 #

async_to_sync会同步执行异步的group_send，可以通过try-except块捕获调用过程中可能出现的错误（比如通道层连接失败、参数格式错误等），同时添加日志记录来确认执行状态：

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

logger = logging.getLogger(__name__)

try:
    channel_layer = get_channel_layer()
    async_to_sync(channel_layer.group_send)(
        f"{mychannelname}",
        {
            "type": "chat_message",
            "message": "send you"
        }
    )
    logger.info(f"消息已成功发送到组 {mychannelname}")
except Exception as e:
    logger.error(f"发送消息到组 {mychannelname} 失败: {str(e)}")
    # 可根据业务需求添加重试、告警等逻辑

2. 接收端ACK确认 #

如果需要确保消息被客户端实际接收处理，可以在消息中添加唯一标识，让consumer处理后反馈确认信号：

发送端修改：

import uuid

# 生成唯一消息ID
message_id = str(uuid.uuid4())
async_to_sync(channel_layer.group_send)(
    f"{mychannelname}",
    {
        "type": "chat_message",
        "message": "send you",
        "message_id": message_id
    }
)
# 可通过数据库/缓存等待ACK，设置超时逻辑

Consumer的chat_message方法修改：

async def chat_message(self, event):
    print("someone call chat_message")
    message = event['message']
    message_id = event.get('message_id')
    
    # 发送消息给客户端
    await self.send(text_data=json.dumps({
        'message': message,
        'message_id': message_id
    }))
    
    # 记录消息已送达（示例：写入数据库）
    if message_id:
        await self.record_message_ack(message_id)

@database_sync_to_async
def record_message_ack(self, message_id):
    # 自行实现数据库更新逻辑，比如标记消息状态为已送达
    from myapp.models import MessageLog
    MessageLog.objects.filter(id=message_id).update(status="delivered")

二、获取连接到组的客户端数量 #

Channels没有内置的组内成员计数API，需要自行维护在线人数，推荐用Redis（和Channels通道层共用实例，支持原子操作）或数据库实现：

1. Redis维护在线计数 #

修改Consumer的connect和disconnect方法：

import json
import redis.asyncio as redis

from channels.db import database_sync_to_async
from channels.generic.websocket import AsyncWebsocketConsumer

# 初始化Redis客户端（和Channels通道层配置一致）
redis_client = redis.from_url("redis://localhost:6379/0")

class ChatConsumer(AsyncWebsocketConsumer):
    
    async def connect(self):
        self.room_group_name = self.scope["url_route"]["kwargs"]["room_name"]

        await self.channel_layer.group_add(
            self.room_group_name,
            self.channel_name
        )
        await self.accept()
        
        # 原子递增组在线人数
        await redis_client.hincrby("room_online_counts", self.room_group_name, 1)
        
        # 给客户端返回当前在线人数
        online_count = await redis_client.hget("room_online_counts", self.room_group_name)
        await self.send(text_data=json.dumps({
            'channel_name': self.channel_name,
            'online_count': int(online_count) if online_count else 0
        }))

    async def disconnect(self, close_code):
        print("some disconnect")
        await self.channel_layer.group_discard(
            self.room_group_name,
            self.channel_name
        )
        
        # 原子递减组在线人数
        await redis_client.hincrby("room_online_counts", self.room_group_name, -1)
        # 计数为0时删除键节省空间
        count = await redis_client.hget("room_online_counts", self.room_group_name)
        if int(count) <= 0:
            await redis_client.hdel("room_online_counts", self.room_group_name)

    # 其他原有方法保持不变...
    
    # 新增静态方法，供外部调用获取在线人数
    @classmethod
    async def get_online_count(cls, room_group_name):
        count = await redis_client.hget("room_online_counts", room_group_name)
        return int(count) if count else 0

外部获取在线人数：

from asgiref.sync import async_to_sync
from myapp.consumers import ChatConsumer

# 同步获取指定组的在线人数
online_count = async_to_sync(ChatConsumer.get_online_count)(mychannelname)
print(f"组 {mychannelname} 在线人数: {online_count}")

2. 数据库维护计数 #

如果不用Redis，可通过数据库表存储计数，注意使用原子操作避免并发问题：

先创建模型：

from django.db import models

class RoomOnlineCount(models.Model):
    room_group_name = models.CharField(max_length=255, unique=True)
    count = models.IntegerField(default=0)
    
    class Meta:
        verbose_name = "房间在线人数"
        verbose_name_plural = verbose_name

修改Consumer方法：

async def connect(self):
    # ... 原有连接逻辑 ...
    
    # 原子递增计数
    await database_sync_to_async(self.increment_count)()
    
async def disconnect(self, close_code):
    # ... 原有断开逻辑 ...
    
    # 原子递减计数
    await database_sync_to_async(self.decrement_count)()

def increment_count(self):
    obj, created = RoomOnlineCount.objects.get_or_create(room_group_name=self.room_group_name)
    RoomOnlineCount.objects.filter(id=obj.id).update(count=models.F('count') + 1)

def decrement_count(self):
    try:
        obj = RoomOnlineCount.objects.get(room_group_name=self.room_group_name)
        if obj.count > 0:
            RoomOnlineCount.objects.filter(id=obj.id).update(count=models.F('count') - 1)
        if obj.count <= 1:
            obj.delete()
    except RoomOnlineCount.DoesNotExist:
        pass

注意事项 #

• 使用Redis时，确保Channels通道层已配置为RedisBackend，可共用同一个Redis实例
• 数据库计数必须用models.F做原子更新，避免并发场景下计数错误
• 若需精确统计，可结合心跳机制清理客户端异常断开的无效连接

内容的提问来源于stack exchange，提问作者whitebear