Django Channels启动异常：本地Windows11正常，同事Windows11及Debian服务器报错

最近遇到一个特别奇怪的问题：我自己的Windows11电脑上Django Channels项目运行完全正常，但同事的同款Windows11电脑，还有我专门用来测试的Debian Linux服务器上，一启动就报错。

所有设备的代码都是通过Git同步的，虚拟环境和依赖库也完全一致（用pip freeze核对过），Python版本除了Linux服务器是3.11.2，我和同事都是3.12.1——但同事和我版本一样也报错，所以应该不是版本的锅。

我在StackOverflow和其他网站都找不到类似的错误案例，实在搞不懂为什么只有我自己的电脑能正常运行！

一开始得到的错误信息非常模糊：

Traceback (most recent call last):
  File "/usr/lib/python3.11/threading.py", line 1038, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.11/threading.py", line 975, in run
    self._target(*self._args, **self._kwargs)
  File "/srv/<redacted>/venv/lib/python3.11/site-packages/django/utils/autoreload.py", line 64, in wrapper
    fn(*args, **kwargs)
  File "/srv/<redacted>/venv/lib/python3.11/site-packages/daphne/management/commands/runserver.py", line 128, in inner_run
    application=self.get_application(options),
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/srv/<redacted>/venv/lib/python3.11/site-packages/daphne/management/commands/runserver.py", line 153, in get_application
    return ASGIStaticFilesHandler(get_default_application())
                                  ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/srv/<redacted>/venv/lib/python3.11/site-packages/daphne/management/commands/runserver.py", line 31, in get_default_application
    raise ImproperlyConfigured("Cannot import ASGI_APPLICATION module %r" % path)
django.core.exceptions.ImproperlyConfigured: Cannot import ASGI_APPLICATION module 'config.asgi'

经过进一步调试，才找到真正的崩溃原因：

ImportError: cannot import name 'DEFAULT_CHANNEL_LAYER' from 'channels'

环境对比 #

• 我的环境：Windows11 - Python 3.12.1（正常运行）
• 同事环境：Windows11 - Python 3.12.1（报错）
• Linux服务器（上述错误栈来自这里）：Debian - Python 3.11.2（报错）

补充几个关键细节：

• 我们的Django Channels是全新安装的，我怀疑问题出在配置或环境初始化上
• 项目还用到了Django Rest Framework、Django Tenants & Django Tenant User这些核心库
• 只有用Daphne ASGI服务器启动时才会报错，用默认WSGI完全没问题
• 只有在导入Consumer的时候才会崩溃，不导入Consumer就不会触发错误

相关代码 #

asgi.py

import os

from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
from django.core.asgi import get_asgi_application
from django.urls import path
from config.middleware import TokenAuthMiddleware


os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings')
# Initialize Django ASGI application early to ensure the AppRegistry
# is populated before importing code that may import ORM models.
django_asgi_app = get_asgi_application()

from apps.chats.consumers import MyWebSocketConsumer

application = ProtocolTypeRouter({
    "http": django_asgi_app,
    "websocket": TokenAuthMiddleware(
        URLRouter(
            [
                path("ws/test/", MyWebSocketConsumer.as_asgi()),
            ]
        )
    ),
})

config/middleware.py

from django.contrib.auth.models import AnonymousUser
from rest_framework.authtoken.models import Token
from channels.db import database_sync_to_async
from channels.middleware import BaseMiddleware

@database_sync_to_async
def get_user(token_key):
    try:
        token = Token.objects.get(key=token_key)
        return token.user
    except Token.DoesNotExist:
        return AnonymousUser()

class TokenAuthMiddleware(BaseMiddleware):
    def __init__(self, inner):
        super().__init__(inner)

    async def __call__(self, scope, receive, send):
        try:
            token_key = dict(scope['headers']).get(b'authorization').decode('ascii').split(" ")[1]
        except ValueError:
            token_key = None
        print("test3")
        print(token_key)
        scope['user'] = AnonymousUser() if token_key is None else await get_user(token_key)
        return await super().__call__(scope, receive, send)

apps/chats/consumers.py

import json
from channels.generic.websocket import JsonWebsocketConsumer
from django_tenants.utils import get_tenant_model

from apps.public.models import Domain
from django_tenants.utils import tenant_context

Tenant = get_tenant_model()

class MyWebSocketConsumer(JsonWebsocketConsumer):    
    def connect(self):
        user = self.scope['user']
        
        tenant_domain = dict(self.scope['headers']).get(b'host').decode('ascii').split(":")[0]
        tenant_domain_obj = Domain.objects.get(domain=tenant_domain)
        tenant = Tenant.objects.filter(domains=tenant_domain_obj).first()
        
        with tenant_context(tenant):
            if user.is_authenticated:
                user_chats = user.chats_followed.all() 
                                
                self.groups = [f"chat_{chat.id}" for chat in user_chats]

                # Subscribe the WebSocket connection to all the user's chat groups
                for group_name in self.groups:
                    self.channel_layer.group_add(group_name, self.channel_name)

                # Accept the WebSocket connection
                self.accept()

    def disconnect(self, close_code):
        # Remove the WebSocket connection from the group
        for group_name in self.groups:
            self.channel_layer.group_discard(group_name, self.channel_name)

    def receive_json(self, content):
        # This is called when a message is received from the client
        message = content.get("message", "")
        chat_id = content.get("chat_id")
        
        print(message)

        # Broadcast the message to the specific chat group
        if chat_id:
            group_name = f"chat_{chat_id}"
            self.channel_layer.group_send(
                group_name,
                {
                    "type": "chat_message",
                    "chat_id": chat_id,
                    "message": message,
                }
            )

    def chat_message(self, event):
        message = event['message']

        # Send the message to the WebSocket
        self.send(text_data=json.dumps({"message": message}))

我已经排查过以下内容，结果都是一致的：

• 环境变量配置正常
• Python版本核对无误
• 安装的依赖包完全匹配
• 操作系统版本也检查过

实在搞不懂为什么只有我自己的电脑能正常运行，到底哪里出问题了？

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