Django Channels Echo Server连接失败问题排查求助

以下是针对你遇到的not found:/ws/及WebSocket连接失败问题的具体排查方向：

• URL路由配置问题

  1. 检查项目根urls.py是否正确引入WebSocket路由，确保存在类似代码：

     from django.urls import path, include
     
     urlpatterns = [
         # 其他HTTP路由
         path('ws/', include('your_app.routing')),
     ]
     

  2. 确认应用的routing.py中，websocket_urlpatterns正确映射路径到消费者：

     from django.urls import path
     from . import consumers
     
     websocket_urlpatterns = [
         path('', consumers.EchoConsumer.as_asgi()),  # 对应根路由的/ws/路径
     ]
     

  注意不要在应用路由里重复添加ws/，避免路径冲突。

• ASGI应用配置错误

  1. 检查项目根asgi.py是否正确配置ProtocolTypeRouter，示例正确配置：

     import os
     from django.core.asgi import get_asgi_application
     from channels.routing import ProtocolTypeRouter, URLRouter
     from channels.auth import AuthMiddlewareStack
     import your_app.routing
     
     os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'your_project.settings')
     
     application = ProtocolTypeRouter({
         "http": get_asgi_application(),
         "websocket": AuthMiddlewareStack(
             URLRouter(
                 your_app.routing.websocket_urlpatterns
             )
         ),
     })
     

  2. 确保settings.py中已添加'channels'到INSTALLED_APPS，且设置ASGI_APPLICATION = 'your_project.asgi.application'。

• 服务器启动方式错误
  必须使用支持ASGI的命令启动服务器，不能依赖传统WSGI启动方式：

  • 使用Daphne启动：

    daphne your_project.asgi:application
    

  • 或使用Django自带命令（Django 3.0+支持）：

    python manage.py runserver
    

• 消费者类实现问题
  检查消费者类是否正确实现必要方法：

  • 同步消费者示例：

    from channels.generic.websocket import WebSocketConsumer
    from channels.exceptions import StopConsumer
    
    class EchoConsumer(WebSocketConsumer):
        def connect(self):
            self.accept()  # 必须调用accept()完成连接
    
        def receive(self, text_data):
            self.send(text_data=text_data)
    
        def disconnect(self, close_code):
            raise StopConsumer()
    

  • 异步消费者需用async def定义方法，并await异步操作：

    from channels.generic.websocket import AsyncWebsocketConsumer
    
    class EchoConsumer(AsyncWebsocketConsumer):
        async def connect(self):
            await self.accept()
    
        async def receive(self, text_data):
            await self.send(text_data=text_data)
    

• 其他潜在问题

  • 检查settings.py的ALLOWED_HOSTS是否包含127.0.0.1，避免主机验证失败。
  • 暂时禁用非必要中间件，排查是否有中间件拦截WebSocket请求。

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