AWS上Django Channels接收消息时出现msgpack.ExtraData反序列化错误

问题描述 #

使用Django 4.2.6 + Django Channels 4.0.0 + Channels-redis 4.1.0在AWS Beanstalk Linux 2023环境开发Web应用时，WebSocket消费者接收客户端消息时抛出反序列化错误：msgpack.exceptions.ExtraData: unpack(b) received extra data。

环境背景 #

• 框架版本：Django 4.2.6、Django Channels 4.0.0、Channels-redis 4.1.0，部署在AWS Beanstalk Linux 2023
• 架构：负载均衡器转发HTTP到WSGI服务器（8000端口，运行正常），WSS流量转发到ASGI服务器（Daphne）处理异步请求；安全组配置正确，流量可正常通行；Elastic Redis缓存服务运行正常，数据读写无问题
• Channels配置：最初使用channels_redis.core.RedisChannelLayer时连接无法保持（客户端连接1秒内断开），切换为channels_redis.pubsub.RedisPubSubChannelLayer后该问题解决，已在settings.py中配置Redis作为Channels后端

预期行为 #

消费者可正常处理传入消息，但当前存在消息反序列化失败问题。

已尝试的解决方法 #

• 验证通过Channels发送的数据格式正确
• 尝试多种不同的数据格式，包括硬编码发送消息
• 检查Django Channels所有配置项，确认配置正确

相关代码与错误栈 #

配置与代码片段 #

# settings.py 中的CHANNEL_LAYERS配置
CHANNEL_LAYERS = {
    'default': {
        'BACKEND': 'channels_redis.pubsub.RedisPubSubChannelLayer',
        'CONFIG': {
            "hosts": [('cacheback-mbxa32.serverless.euw3.cache.amazonaws.com', 6379)],
        },
    },
}

# asgi.py
import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
import appProject.routing
import logging

logger = logging.getLogger(__name__)
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'appProject.settings')
logger.debug("进入ASGI")
application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": AuthMiddlewareStack(
        URLRouter(
            appProject.routing.websocket_urlpatterns
        )
    ),
})

# 消费者的receive方法
async def receive(self, text_data):
    try:
        text_data_json = json.loads(text_data)
        message = text_data_json['message']
        
        # 发送消息到聊天组
        await self.channel_layer.group_send(
            self.room_group_name,
            {
                'type': 'chat_message',
                'message': message
            }
        )
    except Exception as e:
        # 可添加日志记录异常
        pass

错误栈日志 #

2024-01-16 03:46:33,676 No-IP daphne.ws_protocol DEBUG WebSocket ['172.31.9.79', 52020] open and established
2024-01-16 03:46:33,676 No-IP daphne.ws_protocol DEBUG WebSocket ['172.31.9.79', 52020] accepted by application
2024-01-16 03:46:33,676 No-IP daphne.ws_protocol DEBUG Sent WebSocket packet to client for ['172.31.9.79', 52020]
2024-01-16 03:46:33,677 No-IP daphne.ws_protocol DEBUG Sent WebSocket packet to client for ['172.31.9.79', 52020]
2024-01-16 03:46:39,226 No-IP daphne.ws_protocol DEBUG WebSocket incoming frame on ['172.31.9.79', 52020]
2024-01-16 03:46:40,053 No-IP daphne.server ERROR Exception inside application: unpack(b) received extra data.
Traceback (most recent call last):
  File "/var/app/venv/staging-LQM1lest/lib64/python3.11/site-packages/channels/routing.py", line 62, in __call__
    return await application(scope, receive, send)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/var/app/venv/staging-LQM1lest/lib64/python3.11/site-packages/channels/sessions.py", line 47, in __call__
    return await self.inner(dict(scope, cookies=cookies), receive, send)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/var/app/venv/staging-LQM1lest/lib64/python3.11/site-packages/channels/sessions.py", line 263, in __call__
    return await self.inner(wrapper.scope, receive, wrapper.send)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/var/app/venv/staging-LQM1lest/lib64/python3.11/site-packages/channels/auth.py", line 185, in __call__
    return await super().__call__(scope, receive, send)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/var/app/venv/staging-LQM1lest/lib64/python3.11/site-packages/channels/middleware.py", line 24, in __call__
    return await self.inner(scope, receive, send)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/var/app/venv/staging-LQM1lest/lib64/python3.11/site-packages/channels/routing.py", line 116, in __call__
    return await application(
           ^^^^^^^^^^^^^^^^^^
  File "/var/app/venv/staging-LQM1lest/lib64/python3.11/site-packages/channels/consumer.py", line 94, in app
    return await consumer(scope, receive, send)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/var/app/venv/staging-LQM1lest/lib64/python3.11/site-packages/channels/consumer.py", line 58, in __call__
    await await_many_dispatch(
  File "/var/app/venv/staging-LQM1lest/lib64/python3.11/site-packages/channels/utils.py", line 57, in await_many_dispatch
    await task
msgpack.exceptions.ExtraData: unpack(b) received extra data.
2024-01-16 03:46:40,289 No-IP daphne.ws_protocol DEBUG WebSocket closed for ['172.31.9.79', 52020]

问题排查与解决方案 #

可能原因 #

1. Redis残留旧格式数据：切换ChannelLayer后端时，Redis中可能留存了RedisChannelLayer存储的旧数据，新的RedisPubSubChannelLayer用msgpack解析这些数据时出现格式不匹配，触发ExtraData错误。
2. 客户端消息类型不匹配：消费者仅处理文本消息，但客户端可能发送了二进制消息，导致Channels内部用msgpack解析时出错。
3. 版本兼容性问题：Channels-redis 4.1.0与Django Channels 4.0.0的RedisPubSubChannelLayer序列化逻辑存在兼容性冲突。

解决方案建议 #

• 清理Redis缓存：清空Redis中Channels相关的所有键值（生产环境需确认数据可清空，可执行redis-cli FLUSHDB），避免旧数据干扰。
• 修改消费者接收逻辑：同时兼容文本和二进制消息的处理：

  import msgpack
  
  async def receive(self, text_data=None, bytes_data=None):
      message = None
      if text_data:
          text_data_json = json.loads(text_data)
          message = text_data_json['message']
      elif bytes_data:
          # 处理二进制msgpack格式数据
          data = msgpack.unpackb(bytes_data, strict_map_key=False)
          message = data.get('message')
      
      if message:
          await self.channel_layer.group_send(
              self.room_group_name,
              {'type': 'chat_message', 'message': message}
          )
  

• 调整依赖版本：尝试将Channels-redis降级到4.0.0或升级至最新稳定版，验证兼容性问题是否解决。
• 检查负载均衡器配置：确认负载均衡器未对WebSocket帧进行额外封装或修改，避免数据冗余导致解析错误。

内容的提问来源于stack exchange，提问作者Fernando De Palma Madrid