使用Django Channels向同一账号所有设备推送消息遇阻求助

核心问题分析 #

你当前的代码逻辑存在两个关键错误，导致同一账号多设备无法正常接收WebSocket消息：

1. 设备加入Group的逻辑错误：仅在首次创建OnlineUser记录时才将设备加入用户组，后续所有设备连接都未加入组，因此只有第一个设备能接收消息。
2. 设备退出Group的逻辑错误：仅在删除OnlineUser记录时才将设备移出组，导致已断开的设备channel仍留在组内，而在线设备可能不在组内，消息无法正确投递。

Django Channels完全支持同一账号多设备接收消息，Group机制的设计目标就是向所有加入组的channel广播消息，每个设备的WebSocket连接都是独立的channel，只要都加入同一个用户组，就能收到推送。

修正后的代码实现 #

修改NotificationConsumer #

class NotificationConsumer(AsyncWebsocketConsumer):
    def __init__(self, *args, **kwargs):
        super().__init__(args, kwargs)
        self.user = None
        self.pool_name = None

    async def connect(self):
        self.user = self.scope["user"]
        self.pool_name = f"user_id_{self.scope['url_route']['kwargs']['user_id']}"
        await self.accept()

        # 每个设备连接时都加入用户组
        await self.channel_layer.group_add(
            self.pool_name,
            self.channel_name,
        )

        # 更新在线设备计数器
        await self.add_online_user()

    async def disconnect(self, close_code):
        # 每个设备断开时都退出用户组
        await self.channel_layer.group_remove(
            self.pool_name,
            self.channel_name,
        )

        # 更新在线设备计数器
        await self.remove_online_user()

    # 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.pool_name, {"type": "read_message", "message": message})

    # Receive message from room group
    async def notification_message(self, event):
        message = event["message"]
        await self.send(text_data=json.dumps({"message": [message]}, ensure_ascii=False))

    @database_sync_to_async
    def add_online_user(self):
        # 简化计数器更新逻辑，自动创建或递增
        online_user, created = OnlineUser.objects.get_or_create(user=self.user)
        online_user.counter = online_user.counter + 1 if online_user.counter else 1
        online_user.save()

    @database_sync_to_async
    def remove_online_user(self):
        try:
            online_user = OnlineUser.objects.get(user=self.user)
            online_user.counter -= 1
            if online_user.counter <= 0:
                online_user.delete()
            else:
                online_user.save()
        except OnlineUser.DoesNotExist:
            pass

保留OnlineUser模型（无需修改） #

class OnlineUser(models.Model):
    user = models.OneToOneField(User, on_delete=models.CASCADE, related_name="online_user")
    counter = models.IntegerField(null=True, blank=True)

消息发送方法（无需修改，逻辑正确） #

def send_messages(notifications):
    channel_layer = get_channel_layer()
    mark_as_send = []
    for notification in notifications:
        online_user = User.objects.filter(online_user__user=notification.to_user)
        if online_user.exists():
            message = NotificationSerializer(notification).data
            data = {"type": "notification_message", "message": message}
            async_to_sync(channel_layer.group_send)(f"user_id_{notification.to_user.id}", data)
            mark_as_send.append(notification.id)

关键修改说明 #

1. Group操作解耦：将设备加入/退出组的操作与计数器逻辑完全分离，每个WebSocket连接独立处理Group加入和退出，确保所有在线设备都在用户组内。
2. 简化计数器逻辑：使用get_or_create简化在线用户记录的创建，避免空值判断，确保计数器准确反映在线设备数量。
3. 修复退出逻辑：每个设备断开时都从组内移除自身channel，避免无效channel占用组资源，同时正确维护在线计数器。

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