如何在Django服务中实现多WebSocket客户端连接？

架构场景 #

• 主服务：基于django-channels的Django服务，同时提供HTTP和WebSocket服务
• 目标服务：多个Django服务，需与主服务建立多条长连接WebSocket通道，持续收发消息
• 连接URL无法提前预设，需通过主服务发送的运行时负载生成
• 目标服务需同时维护数千条有状态WebSocket连接，且不能影响自身HTTP API的正常响应

当前实现的痛点 #

最初采用同步WebSocket客户端，直接在Django视图中运行会阻塞服务器；改用Celery Worker托管连接后，Worker线程快速耗尽，资源占用过高，仅能临时解决问题。

尝试基于asyncio+websockets实现异步客户端，但遇到以下问题：

1. 连接建立后，代码阻塞在receive_messages循环中，无法执行后续发送逻辑
2. 无法在适配器实例外部主动发送消息，控制权无法返回Django服务进程
3. 无法动态接收新的连接请求，同时维持已有连接的存活与消息处理

现有异步客户端代码 #

import asyncio
import websockets
import json


class WebSocketAdapter:
    def __init__(self, uri):
        self.uri = uri
        self.websocket = None

    async def connect(self):
        try:
            self.websocket = await websockets.connect(self.uri)
            await self.on_connect()
            await self.receive_messages()
        except Exception as e:
            await self.on_error(e)

    async def disconnect(self):
        await self.websocket.close()

    async def send_message(self, message):
        if self.websocket is not None:
            await self.websocket.send(json.dumps(message))
            print("Message Sent")
        else:
            print("WebSocket connection is not established")

    async def receive_messages(self):
        while True:
            try:
                message = await self.websocket.recv()
                await self.on_message(message)
            except websockets.exceptions.ConnectionClosed:
                await self.on_close()
                break
            except Exception as e:
                await self.on_error(e)

    async def on_connect(self):
        print("Connected to WebSocket server.")

    async def on_message(self, message):
        print(f"Received message from server: {message}")

    async def on_error(self, error):
        print("WebSocket error:", error)

    async def on_close(self):
        print("WebSocket connection closed.")


async def main(url):
    websocket_adapter = WebSocketAdapter(url)
    await websocket_adapter.connect()
    # 以下代码永远无法执行
    await websocket_adapter.send_message("Message from client.")


async def test():
    urls = [
        "ws://localhost:8765/conversations/1",
        "ws://localhost:8765/conversations/2",
        "ws://localhost:8765/conversations/3",
    ]
    tasks = [main(url) for url in urls]
    await asyncio.gather(*tasks)


asyncio.run(test())

核心需求 #

1. 目标服务通过HTTP端点接收连接请求，后台异步建立WebSocket连接，同时能持续处理新的HTTP请求
2. 单线程支持数千条WebSocket长连接，资源占用低
3. 保留Celery处理业务数据，但WebSocket连接的管理逻辑需独立于Django和Celery进程
4. 可灵活初始化WebSocket客户端实例，配置自定义回调（连接建立、消息接收、错误、关闭），并能在外部主动发送消息

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