Django+Vue配置WebSocket失败求助：无法连接指定WS地址

我是WebSocket新手，正在配置WebSocket以实时展示Post模型的新条目，但运行后控制台提示连接失败，相关代码及错误信息如下，请帮忙排查问题。

Post模型代码 #

class Post(models.Model):
    title = models.CharField(max_length=200, unique=True)
    content = models.TextField()
    status = models.IntegerField(choices=STATUS, default=0)
    author = models.ForeignKey(
        User,
        related_name="blog_posts",
        on_delete=models.CASCADE,
    )
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)

consumers.py代码 #

class PostConsumer(ListModelMixin, GenericAsyncAPIConsumer):

    queryset = Post.objects.all()
    serializer_class = PostSerializer
    permissions = (permissions.AllowAny,)

    async def connect(self, **kwargs):
        await self.model_change.subscribe()
        await super().connect()

    @model_observer(Post)
    async def model_change(self, message, observer=None, **kwargs):
        await self.send_json(message)

    @model_change.serializer
    def model_serialize(self, instance, action, **kwargs):
        return dict(data=PostSerializer(instance=instance).data, action=action.value)

前端Vue处理WebSocket的核心代码 #

<script>
  vueApp = new Vue({
    el: "#app",
    delimiters: ["[[", "]]"],
    data() {
      return {
        posts: [],
      };
    },
  });

  var ws = new WebSocket("ws://localhost:8001/ws/");
  console.log(this.posts, "CHECKING WEBSOCKETS")
  console.log(ws)
  ws.onopen = function (e) {
    ws.send(
      JSON.stringify({
        action: "list",
        request_id: new Date().getTime(),
      })
    );
  };

  ws.onmessage = function (e) {
    allData = JSON.parse(e.data);
    if (allData.action === "list") {
      vueApp.$data.posts = allData.data;
      vueApp.$forceUpdate();
    } else if (allData.action === "create") {
      vueApp.$data.posts.push(allData.data);
    }
  };
</script>

错误信息 #

WebSocket connection to 'ws://localhost:8001/ws/' failed:

排查方向： #

• 检查Channels基础配置

  • 确保channels和channels-redis（或其他通道层依赖）已安装，且channels添加到INSTALLED_APPS中。
  • 确认项目根目录的asgi.py已替换为Channels的ASGI配置，示例：

    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
            )
        ),
    })
    

  • 在settings.py中配置通道层，以Redis为例：

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

• 确认WebSocket路由配置

  • 在应用目录下创建routing.py，定义WebSocket路由指向PostConsumer：

    from django.urls import path
    from . import consumers
    
    websocket_urlpatterns = [
        path('ws/', consumers.PostConsumer.as_asgi()),
    ]
    

• 检查服务运行方式

  • 必须使用ASGI服务器（如daphne）启动项目，而非默认的WSGI runserver。启动命令示例：daphne your_project.asgi:application --port 8001，确保端口与前端连接的8001一致。

• 跨域问题排查

  • 如果前端页面运行在其他端口（如8000），需配置Django跨域支持：安装django-cors-headers，添加到INSTALLED_APPS，并在settings.py中设置CORS_ALLOW_ALL_ORIGINS = True（测试环境使用，生产环境需指定允许的域名）。

• 消费者代码细节检查

  • 确认PostSerializer已正确定义，能正常序列化Post模型字段。
  • 检查消费者的导入是否正确，比如model_observer、GenericAsyncAPIConsumer等依赖的导入路径是否符合使用的Channels版本。

内容的提问来源于stack exchange，提问作者Mike D Hovhannisyan