Django Channels多窗口场景下group_send推送当前活跃窗口延迟问题

问题产生原因 #

1. 核心原因是Async Consumer 中存在未包装的同步阻塞代码：你在websocket_receive方法里标注的#....部分业务代码...大概率包含同步IO操作（比如Django ORM查询、文件读写、调用同步第三方接口等），而Async Consumer的所有逻辑都运行在同一个协程事件循环中，同步阻塞代码会卡住当前Consumer的事件调度，导致当前Consumer无法及时处理group_send推送过来的master_broadcast事件。
2. 其他窗口不受影响的原因是：每个WebSocket连接对应独立的Consumer实例，事件循环互不干扰，所以其他窗口能正常收到master分组的推送。
3. 后续user分组推送无延迟的原因是：阻塞的业务代码执行完成后，事件循环恢复正常调度，此时user分组的推送事件刚好进入队列，就会被立即处理。
4. 额外排查点：如果本地使用的是默认的InMemoryChannelLayer，也可能存在低概率的事件调度优先级问题，该实现仅适合本地测试使用。

修复方案 #

1. 包装所有同步IO操作 #

所有位于Async Consumer中的同步阻塞逻辑，都需要用database_sync_to_async包装，避免阻塞事件循环，示例修改如下：

from channels.db import database_sync_to_async
import json

# 封装原来的同步业务逻辑
@database_sync_to_async
def get_biz_data(user):
    # 示例：原来的ORM查询、业务计算逻辑都写在这里
    # messages_all = list(Message.objects.all().values("content", "username"))
    # messages_user = list(Message.objects.filter(user=user).values("content", "send_time"))
    return messages_all, messages_user

async def websocket_receive(self, event):
    user = self.scope['user']
    user_group = "user_" + str(user.id)
    
    # 替换原来的同步业务代码，用await调用包装后的方法
    messages_all, messages_user = await get_biz_data(user)

    await self.channel_layer.group_send(
        "master",{
            'type': 'master.broadcast',
            'text': json.dumps(messages_all)
        }
    )
    
    await self.channel_layer.group_send(
        user_group,{
            'type': 'user.broadcast',
            'text': json.dumps(messages_user)
        }
    )

2. 主动让渡事件循环执行权（可选） #

如果你的业务逻辑没有IO操作，但是计算量较大，可以在group_send之后手动添加await asyncio.sleep(0)，主动让出事件循环的执行权，让待处理的广播事件优先执行：

import asyncio

async def websocket_receive(self, event):
    # ... 其他代码 ...
    await self.channel_layer.group_send(
        "master",{
            'type': 'master.broadcast',
            'text': json.dumps(messages_all)
        }
    )
    # 加这行主动让出事件循环
    await asyncio.sleep(0)
    
    await self.channel_layer.group_send(
        user_group,{
            'type': 'user.broadcast',
            'text': json.dumps(messages_user)
        }
    )

3. 替换为Redis ChannelLayer #

生产环境建议替换掉测试用的内存ChannelLayer，改用Redis实现：
首先安装依赖：

pip install channels-redis==3.4.1

然后修改settings.py配置：

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

4. 调整加组逻辑（优化项） #

当前加组逻辑存在未登录用户也被加入master组的问题，可以调整顺序避免无效加组：

async def websocket_connect(self, event):
    user = self.scope['user']
    # 先判断登录状态，未登录可直接拒绝连接，或者根据业务逻辑处理
    if not user.is_authenticated:
        await self.send({"type": "websocket.close"})
        return

    await self.send({
        "type": "websocket.accept"
    })
    await self.channel_layer.group_add("master", self.channel_name)
    user_group = "user_" + str(user.id)
    await self.channel_layer.group_add(user_group, self.channel_name)        

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