使用Django Channels搭建通知系统遇UserWarning问题求助

问题根源 #

警告的核心原因是：同步消费者（WebsocketConsumer）的channel_layer接口已经是同步适配后的版本，你却用async_to_sync包裹了同步方法，导致async_to_sync接收到非异步标记的可调用对象，触发警告。同时代码中还有两处功能逻辑问题，一并修正：

分步解决方案 #

1. 修正消费者代码（移除多余的async_to_sync） #

修改consumers.py，移除async_to_sync包裹，并确保事件处理方法名和信号中的type字段一致，补充断开连接时的group清理逻辑：

from channels.generic.websocket import WebsocketConsumer

class StoreConsumer(WebsocketConsumer):
    def connect(self):
        self.store_slug = self.scope['url_route']['kwargs']['store_slug']
        self.store_group_name = f'store_{self.store_slug}'
        # 同步消费者直接调用channel_layer的同步方法，无需async_to_sync
        self.channel_layer.group_add(
            self.store_group_name,
            self.channel_name
        )

        self.accept()

    def disconnect(self, close_code):
        # 断开连接时从group移除当前channel，避免内存泄漏
        self.channel_layer.group_discard(
            self.store_group_name,
            self.channel_name
        )

    # 方法名必须与信号中发送的type字段完全一致：store_notification
    def store_notification(self, event):
        message = event['message']
        self.send(text_data=message)

2. 修正信号中的Group名称匹配 #

消费者用store_{store_slug}作为group名称，但信号中用了store_{instance.store.id}，两者不匹配会导致消息无法送达。统一为slug（或统一为id，保持一致即可）：

from django.dispatch import receiver
from channels.layers import get_channel_layer
from asgiref.sync import async_to_sync
from stores.models import Notification
from django.db.models.signals import post_save

@receiver(post_save, sender=Notification)
def send_notification(sender, instance, **kwargs):
    channel_layer = get_channel_layer()
    # 使用store的slug生成group名称，与消费者逻辑对齐
    async_to_sync(channel_layer.group_send)(
        f'store_{instance.store.slug}',
        {
            'type': 'store_notification',  # 对应消费者中的处理方法名
            'message': instance.message
        }
    )

3. 移除ASGI文件中提前初始化的channel_layer #

asgi.py中提前调用get_channel_layer()会在应用完全初始化前获取channel layer，可能引发异常，直接删除该行：

import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from stores.urls import websocket_urlpatterns

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'neo.settings')

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": AuthMiddlewareStack(
        URLRouter(
            websocket_urlpatterns
        ),
    )
})

# 删除此行：避免提前初始化channel layer
# channel_layer = get_channel_layer()

若使用AsyncWebsocketConsumer的适配方案 #

如果改用异步消费者，逻辑需要调整为异步风格，无需async_to_sync包裹channel layer方法：

from channels.generic.websocket import AsyncWebsocketConsumer

class StoreConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        self.store_slug = self.scope['url_route']['kwargs']['store_slug']
        self.store_group_name = f'store_{self.store_slug}'
        await self.channel_layer.group_add(
            self.store_group_name,
            self.channel_name
        )

        await self.accept()

    async def disconnect(self, close_code):
        await self.channel_layer.group_discard(
            self.store_group_name,
            self.channel_name
        )

    async def store_notification(self, event):
        message = event['message']
        await self.send(text_data=message)

此时信号中的async_to_sync仍需保留，因为Django信号默认在同步环境执行。

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