Django Channels中WebSocket连接失败与意外关闭问题求助

我在使用Django Channels搭建的Django项目里，WebSocket连接的建立和维持遇到了问题。我做了一个通知系统，想用WebSocket给连接的客户端广播消息，但现在碰到两个错误：

• "WebSocket connection to 'ws://127.0.0.1:8000/ws/notification/broadcast/' failed"
• "Chat socket closed unexpectedly"

下面是我的代码，麻烦帮忙看看哪里出问题了：

routing.py #

from django.urls import re_path  
from . import consumers  
  
websocket_urlpatterns = [  
   re_path(r'ws/notification/(?P<room_name>\w+)/$', consumers.NotificationConsumer.as_asgi()),  
]

views.py #

def notification_view(request):  
   return render(request,'person/notification.html',{'room_name': "broadcast"})

settings.py #

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

# Celery settings
CELERY_BROKER_URL = 'redis://localhost:6379/0'
CELERY_ACCEPT_CONTENT = ['json']
CELERY_TASK_SERIALIZER = 'json'

consumer_context_processor.py #

from notification_app.models import BroadcastNotification  
def notifications(request):  
   allnotifications = BroadcastNotification.objects.all()  
   return {'notifications': allnotifications}

models.py #

class BroadcastNotification(models.Model):  
   message = models.TextField()  
   notification_image=models.ImageField(upload_to="notification",default="notification.jpg",blank=True,null=True)  
   notification_link = models.URLField(max_length=10000, help_text="Add a valid URL",blank=True,null=True)  
   broadcast_on = models.DateTimeField()  
   sent = models.BooleanField(default=False)  
  
   class Meta:  
      ordering = ['-broadcast_on']  
  
@receiver(post_save, sender=BroadcastNotification)  
def notification_handler(sender, instance, created, **kwargs):  
   # call group_send function directly to send notificatoions or you can create a dynamic task in celery beat  
   if created:  
      schedule, created = CrontabSchedule.objects.get_or_create(hour = instance.broadcast_on.hour, minute = instance.broadcast_on.minute, day_of_month = instance.broadcast_on.day, month_of_year = instance.broadcast_on.month)  
      task = PeriodicTask.objects.create(crontab=schedule, name="broadcast-notification-"+str(instance.id), task="notifications_app.tasks.broadcast_notification", args=json.dumps((instance.id,)))

task.py #

@shared_task(bind = True)  
def broadcast_notification(self, data):  
   print(data)  
   try:  
      notification = BroadcastNotification.objects.filter(id = int(data))  
      if len(notification)>0:  
        notification = notification.first()  
        channel_layer = get_channel_layer()  
        loop = asyncio.new_event_loop()  
        asyncio.set_event_loop(loop)  
        loop.run_until_complete(channel_layer.group_send(  
           "notification_broadcast",  
           {  
              'type': 'send_notification',  
              'message': json.dumps(notification.message),  
           }))  
        notification.sent = True  
        notification.save()  
        return 'Done'  
  
      else:  
        self.update_state(  
           state = 'FAILURE',  
           meta = {'exe': "Not Found"}  
        )  
  
        raise Ignore()  
  
   except:  
      self.update_state(  
           state = 'FAILURE',  
           meta = {  
                'exe': "Failed"  
                # 'exc_type': type(ex).__name__,  
                # 'exc_message': traceback.format_exc().split('\n')  
                # 'custom': '...'  
              }  
        )  
  
      raise Ignore()

Frontend (JavaScript) #

{{ room_name|json_script:"room-name" }}  
   <script>  
      const roomName = JSON.parse(document.getElementById('room-name').textContent);  

      const notificationSocket = new WebSocket(  
        'ws://'  
        + window.location.host  
        + '/ws/notification/'  
        + roomName  
        + '/'  
      );  

      notificationSocket.onmessage = function(e) {  
        const data = JSON.parse(e.data);  
        //document.querySelector('#chat-log').value += (data.message + '\n');  
        console.log(data);  
        document.getElementById("notifications-dropdown").innerHTML = "<li class='dropdown-item'>" + data + "</li><hr class='dropdown-divider'>" + document.getElementById("notifications-dropdown").innerHTML;  
        document.getElementById("notification-badge").innerHTML = parseInt(document.getElementById("notification-badge").innerHTML) + 1;  
      };  

      notificationSocket.onclose = function(e) {  
        console.error('Chat socket closed unexpectedly');  
      };  
   </script>

consumers.py #

class NotificationConsumer(AsyncWebsocketConsumer):  
   async def connect(self):  
      self.room_name = self.scope['url_route']['kwargs']['room_name']  
      self.room_group_name = 'notification_%s' % self.room_name  

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

      await self.accept()  

   async def disconnect(self, close_code):  
      # Leave room group  
      await self.channel_layer.group_discard(  
        self.room_group_name,  
        self.channel_name  
      )  


   # Receive message from room group  
   async def send_notification(self, event):  
      message = json.loads(event['message'])  

      # Send message to WebSocket  
      await self.send(text_data=json.dumps(message))

asgi.py #

"""
ASGI config for ecomprj 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/5.0/howto/deployment/asgi/
"""

import os
import django
from channels.routing import ProtocolTypeRouter, URLRouter
from django.core.asgi import get_asgi_application
from channels.auth import AuthMiddlewareStack
from notification_app.routing import websocket_urlpatterns  # Ensure this import is correct

# Set the default settings module for the 'django' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'ecomprj.settings')
django.setup()

# Define the ASGI application
application = ProtocolTypeRouter({
    "http": get_asgi_application(),  # Handle HTTP requests
    "websocket": AuthMiddlewareStack(  # Handle WebSocket requests
        URLRouter(
            websocket_urlpatterns  # Use the WebSocket URL patterns
        )
    )
})

# Note: The second assignment to `application` has been removed

我已经跟着Django Channels官方文档和示例做了，但还是遇到这些问题，有没有大佬能帮忙找出问题所在并给出解决方案？

问题排查与解决方案 #

我帮你梳理了代码里的几个关键问题，逐个来看：

1. Consumer中断开连接方法错误 #

在consumers.py的disconnect方法里，你用了group_discard，但Channels的正确方法是group_remove，这个错误会导致连接断开时出现异常，进而引发“socket closed unexpectedly”的错误：

修改consumers.py的disconnect方法：

async def disconnect(self, close_code):
    # Leave room group
    await self.channel_layer.group_remove(
        self.room_group_name,
        self.channel_name
    )

2. Celery任务中调用异步Channel Layer的方式错误 #

在task.py里，你手动创建事件循环来调用channel_layer.group_send，这在同步环境（Celery任务是同步的）里容易出问题。正确的做法是用async_to_sync来包装异步方法：

首先导入async_to_sync：

from asgiref.sync import async_to_sync

然后修改任务中的调用部分：

# 替换原来的loop相关代码
async_to_sync(channel_layer.group_send)(
    "notification_broadcast",
    {
        'type': 'send_notification',
        'message': json.dumps(notification.message),
    }
)

3. 检查Redis服务与依赖 #

确保你已经安装了channels-redis包：

pip install channels-redis

同时确认Redis服务正在本地运行（端口6379），可以用redis-cli ping来测试，如果返回PONG说明正常。

4. 验证ASGI服务器启动方式 #

开发环境下，建议用Daphne启动ASGI应用，而不是默认的runserver（虽然runserver也支持，但Daphne更稳定）：

daphne ecomprj.asgi:application

或者用：

python manage.py runserver

但要确保你的asgi.py配置正确，目前看起来你的asgi.py是没问题的。

5. 认证中间件的影响 #

如果你的通知系统不需要用户认证，那么AuthMiddlewareStack可能会拒绝未登录用户的WebSocket连接。可以尝试暂时去掉中间件，改成：

# asgi.py中的websocket部分
"websocket": URLRouter(
    websocket_urlpatterns
)

如果这样连接正常了，说明是认证的问题，你可以根据需求选择合适的认证中间件，比如TokenAuthMiddleware或者自定义中间件。

6. 前端WebSocket连接的小优化 #

可以在前端添加错误处理，方便排查问题：

notificationSocket.onerror = function(e) {
    console.error('WebSocket error:', e);
};

测试步骤 #

1. 先修复consumers.py的group_remove问题，重启服务器
2. 修复Celery任务中的async_to_sync调用，重启Celery worker
3. 确认Redis运行正常
4. 访问notification_view页面，查看控制台是否还有连接错误

按照这些步骤修改后，应该能解决大部分问题。如果还有错误，可以查看Django的日志、Celery的日志以及Redis的日志，定位具体的错误信息。

备注：内容来源于stack exchange，提问作者Google User