如何将WebSocket数据异步传递至另一个Python程序

方案1：同一进程内拆分异步任务（适合处理逻辑可嵌入同进程） #

核心用asyncio.Queue做缓冲区，WebSocket客户端负责持续接收消息并写入队列，处理任务从队列取消息处理，两者互不阻塞。

代码示例 #

import asyncio
import json
import websockets

api_data = {
    "type": "subscribe",
    "channel": "v4_orderbook",
    "id": 'BTC-USD',
}

# 创建带缓冲区的异步队列，可根据需求调整maxsize
message_queue = asyncio.Queue(maxsize=1000)

async def ws_receiver(uri):
    async for websocket in websockets.connect(uri):
        await websocket.send(json.dumps(api_data))
        while True:
            message = await websocket.recv()
            try:
                # 非阻塞写入队列，避免阻塞WebSocket接收
                message_queue.put_nowait(message)
            except asyncio.QueueFull:
                # 队列满时可选择丢弃旧消息、记录日志或扩容
                print("消息队列已满，丢弃当前消息")

async def message_processor():
    while True:
        # 阻塞等待队列中的消息
        message = await message_queue.get()
        # 替换为你的实际处理逻辑（比如调用外部程序接口、数据解析等）
        # 模拟耗时处理
        await asyncio.sleep(1)
        print(f"处理完成：{message[:50]}...")
        # 标记任务完成，用于队列的join()方法（可选）
        message_queue.task_done()

async def main():
    # 同时启动接收和处理协程
    await asyncio.gather(
        ws_receiver('wss://indexer.v4testnet.dydx.exchange/v4/ws'),
        message_processor()
    )

asyncio.run(main())

优点 #

• 无需额外依赖，实现简单
• 队列缓冲区保证WebSocket接收不被处理逻辑阻塞
• 同一进程内通信效率高

缺点 #

• 处理逻辑崩溃会连带影响WebSocket客户端
• 不适合处理程序需要独立部署的场景

方案2：跨进程异步通信（适合处理程序独立运行） #

如果必须用独立的Python程序处理数据，推荐用Redis异步队列做中间层，完全解耦两个程序，缓冲区由Redis管理，避免WebSocket客户端阻塞。

步骤1：WebSocket客户端（生产者） #

先安装依赖：pip install aioredis

import asyncio
import json
import websockets
import aioredis

api_data = {
    "type": "subscribe",
    "channel": "v4_orderbook",
    "id": 'BTC-USD',
}

async def ws_producer(uri, redis):
    async for websocket in websockets.connect(uri):
        await websocket.send(json.dumps(api_data))
        while True:
            message = await websocket.recv()
            # 异步写入Redis队列，不会阻塞WebSocket接收
            await redis.lpush("dydx_orderbook_queue", message)
            # 限制队列长度，避免Redis内存占用过高（可选）
            await redis.ltrim("dydx_orderbook_queue", 0, 999)

async def main():
    # 连接本地Redis服务，需提前启动Redis
    redis = await aioredis.from_url("redis://localhost")
    await ws_producer('wss://indexer.v4testnet.dydx.exchange/v4/ws', redis)

asyncio.run(main())

步骤2：数据处理程序（消费者） #

import asyncio
import aioredis

async def message_consumer():
    redis = await aioredis.from_url("redis://localhost")
    while True:
        # 阻塞等待队列中的消息（从队列尾部取出，保证顺序）
        message = await redis.rpop("dydx_orderbook_queue")
        if message:
            message_str = message.decode('utf-8')
            # 替换为你的实际耗时处理逻辑
            await asyncio.sleep(1)
            print(f"处理消息：{message_str[:50]}...")

asyncio.run(message_consumer())

优点 #

• 完全解耦，两个程序独立运行，一个崩溃不影响另一个
• Redis队列支持分布式，可扩展多个处理程序
• 缓冲区容量灵活，可通过Redis配置调整

缺点 #

• 需要额外部署Redis服务
• 跨进程通信效率略低于同进程方案

关键注意点 #

• 无论哪种方案，都要保证WebSocket客户端的消息写入操作是非阻塞/异步非阻塞的，避免拖慢接收流程
• 缓冲区大小要根据消息频率和处理速度合理设置，防止内存溢出或消息丢失
• 队列满时要做容错处理（如丢弃旧消息、报警），避免程序阻塞

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