如何读取StreamReader内容但不消耗其内部缓冲区？

StreamReader是绑定了底层传输事件回调的有状态对象，除了公开的缓冲区和传输层属性外，内部还维护了EOF标记、等待协程、事件循环绑定等私有状态，直接手动复制部分属性创建的新实例无法接收底层传输的新数据通知：当新实例的缓冲区数据不足以满足读取请求时，其内部等待协程永远不会被新数据事件唤醒，最终会被标记为EOF抛出IncompleteReadError。同时同一传输层绑定多个StreamReader实例也会导致数据分发混乱，不建议用复制StreamReader的方式实现需求。

方案1：使用内置peek方法（最简便） #

StreamReader自带的peek()方法原生支持读取缓冲区内容而不消耗数据：

# 最多读取1024字节内容，不会移动缓冲区指针，后续读取操作仍能拿到完整数据
peek_data = await reader.peek(1024)

如果需要读取整行不消耗，可以循环peek判断是否包含换行符：

async def peek_line(reader: StreamReader) -> bytes:
    buf = b""
    while True:
        chunk = await reader.peek(4096)
        if b"\n" in chunk:
            buf += chunk[:chunk.index(b"\n") + 1]
            break
        if not chunk:
            # 已到EOF
            buf += chunk
            break
        buf += chunk
        # 跳过已peek的部分等待新数据
        await reader.read(len(chunk))
    # 将读过的内容塞回缓冲区
    reader._buffer = bytearray(buf) + reader._buffer
    return buf

方案2：读取后回写缓冲区 #

如果需要执行任意读取操作后恢复缓冲区状态，直接将读取到的内容插回原reader的内部缓冲区头部即可：

async def _handle_req(self, reader: StreamReader, writer: StreamWriter):
    # 正常读取需要的内容
    line = await reader.readline()
    # 处理读取到的内容，处理完成后将内容塞回缓冲区
    reader._buffer = bytearray(line) + reader._buffer
    # 后续对reader的读取操作会正常从line的开头开始读取，无数据丢失

上述方案依赖StreamReader的内部_buffer属性，该属性是Python标准库asyncio模块的稳定内部实现，跨版本兼容性有保障，不需要担心后续版本变更导致不可用。
不要尝试创建多个StreamReader实例绑定同一传输层，会导致底层数据分发异常。

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