如何使用Django Channels实现单WebSocket连接下的消息并行处理？

我之前处理过类似的场景，你的问题核心在于AsyncWebsocketConsumer默认会按顺序处理同一连接上的消息——因为每个消息触发的receive协程会被await阻塞，直到当前的耗时任务完成，才会处理下一个消息。要实现并行处理，关键是把耗时任务从receive的同步流程中剥离，放到后台异步执行。

解决方案：用asyncio.create_task异步执行耗时任务 #

你只需要修改receive方法，不要直接await耗时的complicated_answer，而是用asyncio.create_task把它包装成一个后台任务，这样receive能快速返回，立即处理下一个WebSocket消息。

修改后的服务端代码：

import asyncio
from channels.generic.websocket import AsyncWebsocketConsumer

class QuestionConsumer(AsyncWebsocketConsumer):
    async def websocket_connect(self, event):
        await self.accept()

    async def complicated_answer(self, question):
        await asyncio.sleep(3)
        return {
            "What is the Answer to Life, The Universe and Everything?": "42",
            "Why?": "Because.",
        }.get(question, "Don't know")

    async def receive(self, text_data=None, bytes_data=None):
        # 用create_task把耗时任务丢到后台，不阻塞receive处理下一个消息
        asyncio.create_task(self._process_question(text_data))

    async def _process_question(self, question):
        # 单独的方法处理耗时逻辑并发送响应
        answer = await self.complicated_answer(question)
        await self.send(answer)

为什么这样能行？ #

• asyncio.create_task会把_process_question协程加入事件循环，让它和其他任务（比如下一个消息的receive）并行执行。
• 原来的receive方法不再被await complicated_answer阻塞，收到新消息时能立即启动新的后台任务，多个耗时任务就可以同时运行了。

验证结果 #

修改后运行你的客户端代码，就能得到预期的输出：

42
3.0 seconds since first request
Because.
3.0 seconds since first request

额外注意事项 #

• 如果你的complicated_answer涉及数据库操作，一定要用异步数据库驱动（比如asyncpg代替psycopg2），否则同步操作会阻塞整个事件循环，还是无法实现并行。
• 记得给后台任务加上错误处理，避免任务抛出异常导致消费者崩溃：

  async def _process_question(self, question):
      try:
          answer = await self.complicated_answer(question)
          await self.send(answer)
      except Exception as e:
          # 处理异常，比如发送错误响应给客户端
          await self.send(f"Error processing question: {str(e)}")
  

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