aiohttp如何查看WebSocket消息缓冲区？

好问题！aiohttp的客户端WebSocket实现确实在内部维护了一个接收缓冲区，但它并没有提供公开的API来直接访问这个缓冲区——毕竟这属于库的内部实现细节。不过我们可以通过一些变通的方法来实现类似的需求，下面给你两种靠谱的方案：

方案一：包装WebSocket对象，自定义可见缓冲区 #

async for msg in ws: 本质上是循环调用ws.__anext__()，而这个方法内部会调用ws.recv()获取下一条消息。我们可以创建一个包装类，拦截recv()方法，把收到的消息先存到自己维护的缓冲区里，再返回给调用者。这样既不影响原有的循环逻辑，又能随时查看未处理的消息。

import aiohttp
from collections import deque
import asyncio

class WrappedWebSocket:
    def __init__(self, ws):
        self._ws = ws
        self._buffer = deque()  # 我们自定义的可见缓冲区

    async def recv(self):
        # 优先从自定义缓冲区取消息
        if self._buffer:
            return self._buffer.popleft()
        # 缓冲区为空时，调用原始WebSocket的recv方法
        msg = await self._ws.recv()
        return msg

    def peek_buffer(self):
        # 返回当前缓冲区的所有未处理消息（返回副本，避免外部修改内部状态）
        return list(self._buffer)

    # 代理原始WebSocket的所有其他属性和方法，保证功能正常
    def __getattr__(self, name):
        return getattr(self._ws, name)

    # 支持async for循环
    def __aiter__(self):
        return self

    async def __anext__(self):
        msg = await self.recv()
        if msg is None:
            raise StopAsyncIteration
        return msg

# 使用示例
async def main():
    async with aiohttp.ClientSession() as session:
        async with session.ws_connect('wss://example.com') as ws:
            wrapped_ws = WrappedWebSocket(ws)

            # 启动后台任务：持续接收消息并存入自定义缓冲区
            async def background_recv():
                while True:
                    try:
                        msg = await wrapped_ws._ws.recv()
                        wrapped_ws._buffer.append(msg)
                    except (aiohttp.WSServerHandshakeError, asyncio.CancelledError):
                        break
                    except Exception:
                        break

            asyncio.create_task(background_recv())

            # 主循环处理消息，同时可随时查看缓冲区
            while True:
                print("当前未处理消息：", wrapped_ws.peek_buffer())
                msg = await wrapped_ws.recv()
                print("正在处理消息：", msg)

asyncio.run(main())

方案二：手动用asyncio.Queue管理消息 #

放弃使用async for循环，而是自己启动一个后台任务，不断接收WebSocket消息并放入asyncio.Queue中。Queue本身就是一个可见的缓冲区，你可以随时查看它的状态和内容。

import aiohttp
import asyncio

async def main():
    async with aiohttp.ClientSession() as session:
        async with session.ws_connect('wss://example.com') as ws:
            msg_queue = asyncio.Queue()

            # 后台任务：持续接收消息并存入队列
            async def recv_task():
                while True:
                    try:
                        msg = await ws.recv()
                        await msg_queue.put(msg)
                    except Exception:
                        # 连接关闭时放入终止信号
                        await msg_queue.put(None)
                        break

            asyncio.create_task(recv_task())

            # 主逻辑：处理消息+查看缓冲区
            while True:
                # 查看当前队列状态（注意：直接访问_queue是私有属性，仅建议用于调试）
                print("未处理消息数量：", msg_queue.qsize())
                print("未处理消息内容：", list(msg_queue._queue))

                msg = await msg_queue.get()
                if msg is None:
                    break
                print("正在处理消息：", msg)
                msg_queue.task_done()

asyncio.run(main())

注意事项 #

不要直接去访问aiohttp内部的私有缓冲区（比如某些版本中的ws._buffer），因为这些属性属于未公开的实现细节，库版本更新时很可能会修改它们的结构或名称，导致你的代码突然失效。上面的两种方案都是基于aiohttp的公开API实现的，兼容性和稳定性更好。

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