• 游戏开发
• 后端

本篇技术指南以当前仓库内嵌 Python 3.7.3 标准库文档（asyncio-stream.rst）为主体，系统讲解 asyncio 高层 Streams API——open_connection/start_server/StreamReader/StreamWriter的完整用法、参数语义、读写方法与异常处理，并结合仓库源码（streams.py）剖析其底层缓冲、流控与回调机制。读完本文，你将能够在不接触低层 Protocol/Transport 的前提下，用十余行代码编写出 TCP 回声服务器、HTTP 头查询等网络程序，并理解每条 API 背后的实现原理。

Streams 是什么：异步网络 IO 的高层抽象

在 asyncio 的编程模型中，网络 IO 通常有两个层次：

• 低层：Protocol/Transport回调模型，你需要实现connection_made、data_received等回调方法，数据以事件驱动方式被动抵达；
• 高层：Streams。它把低层回调封装成async/await原语，让你像读写普通文件一样收发数据，无需关心回调与传输细节。

当前文档开篇即给出定义：Streams 是用于网络连接的高层、可await的原语，允许在不使用回调或低层协议/传输的情况下发送和接收数据。这一点在源码中得到印证：StreamReaderProtocol类（streams.py#L212-L273）正是「协议到 Stream 的适配器」——它把低层data_received(eof_received)回调翻译成StreamReader.feed_data() / feed_eof()，从而把回调世界桥接到协程世界。

四个顶层函数：建立连接与启动服务

asyncio 模块提供四个顶层协程函数用于创建和使用流：

open_connection：客户端连接入口

签名如下（ssl_handshake_timeout自 Python 3.7 起可用）：

asyncio.open_connection(host=None, port=None, *, loop=None, limit=None, ssl=None, family=0, proto=0, flags=0, sock=None, local_addr=None, server_hostname=None, ssl_handshake_timeout=None)

• 返回的reader与writer分别是StreamReader与StreamWriter实例；
• loop参数可选，当从协程内await该函数时，事件循环总会被自动推断；
• limit决定返回的StreamReader实例的缓冲区大小上限，默认 64 KiB；
• 其余参数原样透传给loop.create_connection（例如ssl、family、sock、local_addr、server_hostname等）。

源码实现（streams.py#L53-L79）展示了它的包装逻辑：先创建StreamReader(limit=limit)，再包装为StreamReaderProtocol，随后调用loop.create_connection(lambda: protocol, host, port, **kwds)拿到传输层，最终组装出StreamWriter并返回(reader, writer)元组。文档注释也坦诚说明：如果想自定义StreamReader或StreamReaderProtocol，直接复制这段代码即可——它本质上只是便利封装。

start_server：服务器启动入口

asyncio.start_server(client_connected_cb, host=None, port=None, *, loop=None, limit=None, family=socket.AF_UNSPEC, flags=socket.AI_PASSIVE, sock=None, backlog=100, ssl=None, reuse_address=None, reuse_port=None, ssl_handshake_timeout=None, start_serving=True)

• client_connected_cb在每次新客户端连接建立时被调用，接收(reader, writer)两个参数；
• 它既可以是普通可调用对象，也可以是协程函数——如果是协程函数，会自动被调度为Task。这一点在源码connection_made中落实（streams.py#L229-L239）：回调返回值若为协程，则通过loop.create_task(res)调度；
• limit默认 64 KiB，控制每个连接的StreamReader缓冲区上限；
• backlog默认 100，对应 socket 的listen()排队长度；reuse_address/reuse_port控制地址复用；start_serving（3.7 新增）为False时只创建服务器而不立即开始服务；
• 其余参数透传给loop.create_server；
• 返回值与loop.create_server一致，即Server对象，可用于停止服务。

在 base_events.py 中可以看到这些参数的底层落实：例如reuse_address在 POSIX（非 Cygwin）平台上默认开启（reuse_address = os.name == 'posix' and sys.platform != 'cygwin'），backlog被用于sock.listen(backlog)，ssl_handshake_timeout则在非 SSL 场景下会被校验拒绝。

Unix socket 变体

asyncio.open_unix_connection(path=None, *, loop=None, limit=None, ssl=None, sock=None, server_hostname=None, ssl_handshake_timeout=None) asyncio.start_unix_server(client_connected_cb, path=None, *, loop=None, limit=None, sock=None, backlog=100, ssl=None, ssl_handshake_timeout=None, start_serving=True)

• 与 TCP 版本行为一致，只是底层改走loop.create_unix_connection/loop.create_unix_server；
• 仅 Unix 平台可用（文档标注.. availability:: Unix.，源码中这两个函数也被if hasattr(socket, 'AF_UNIX')条件包裹）；
• Python 3.7 起path参数可以是 path-like 对象（如pathlib.Path）。

StreamReader：流式读取 API

文档明确指出，不建议直接实例化StreamReader，应通过open_connection/start_server间接获得。类内部维护一个bytearray缓冲区，读取操作本质上是「等数据 → 从缓冲区取走」。

读取方法

细节要点：

• read(0)立即返回空字节；readexactly(0)返回空字节，负数参数抛ValueError（streams.py#L623-L625、L664-L665）；
• readline内部复用readuntil(b'\n')，超限时若缓冲区以换行开头则把完整行移除，否则清空缓冲区并抛ValueError；
• StreamReader还实现了异步迭代协议：async for line in reader:等价于逐行readline()，空字节时抛出StopAsyncIteration结束迭代（streams.py#L690-L697）。

缓冲、限流与并发约束（源码级原理）

• 默认限制 64 KiB：_DEFAULT_LIMIT = 2 ** 16（streams.py#L19），limit必须大于 0，否则构造时抛ValueError；
• 背压（backpressure）：feed_data在缓冲区超过2 * limit时调用transport.pause_reading()暂停底层读取，待消费到limit以内再由_maybe_resume_transport恢复——这是防止内存被无限堆积的关键机制（streams.py#L427-L447）；
• 单等待者约束：_wait_for_data若发现已有协程在等待数据，会抛RuntimeError——即同一时刻只允许一个读协程挂起等待，否则无法判定谁获得下一块数据（streams.py#L449-L475）；
• 暂停时自动恢复：若流因缓冲超限被暂停，_wait_for_data会先resume_reading()再挂起，避免readexactly(n)在n > limit时死锁（streams.py#L465-L469）。

相关异常

• IncompleteReadError(EOFError)：属性partial（EOF 前已读取的字节）与expected（期望的总字节数，未知时为None），由readexactly、readuntil抛出；
• LimitOverrunError：在查找分隔符时缓冲区达到限制而抛出，属性consumed表示已被消费的字节总数，数据仍保留在内部缓冲区中。

两者均定义在 streams.py#L22-L50，并在 asyncio-exceptions.rst 中被单独归档说明。

StreamWriter：流式写入 API

StreamWriter是对Transport的薄封装（streams.py#L276-L284），同样不建议直接实例化。

写入与关闭方法

drain 的流控原理

drain()是 Streams 与低层协议流控衔接的核心（streams.py#L328-L348）：

1. 若关联 reader 已有异常，先抛出该异常；
2. 若传输正在关闭，先await sleep(0)让事件循环有机会触发connection_lost()，否则循环中write(); await drain()会永不感知连接错误；
3. 委托_drain_helper()：若传输未暂停则立即返回；若已暂停，则创建一个 future 挂起，等待resume_writing()或connection_lost()唤醒（streams.py#L147-L209）。

因此规范写法永远是：

writer.write(data) await writer.drain()

完整示例：四个开箱即用的网络程序

以下示例全部来自文档的 Examples 章节，可直接复制运行。

1. 基于 open_connection 的 TCP 回声客户端

import asyncio async def tcp_echo_client(message): reader, writer = await asyncio.open_connection( '127.0.0.1', 8888) print(f'Send: {message!r}') writer.write(message.encode()) data = await reader.read(100) print(f'Received: {data.decode()!r}') print('Close the connection') writer.close() await writer.wait_closed() asyncio.run(tcp_echo_client('Hello World!'))

运行前需要先启动服务端；asyncio.run负责创建事件循环、执行协程并清理资源。

2. 基于 start_server 的 TCP 回声服务器

import asyncio async def handle_echo(reader, writer): data = await reader.read(100) message = data.decode() addr = writer.get_extra_info('peername') print(f"Received {message!r} from {addr!r}") print(f"Send: {message!r}") writer.write(data) await writer.drain() print("Close the connection") writer.close() async def main(): server = await asyncio.start_server( handle_echo, '127.0.0.1', 8888) addr = server.sockets[0].getsockname() print(f'Serving on {addr}') async with server: await server.serve_forever() asyncio.run(main())

要点：handle_echo是协程回调，会被自动调度为 Task（每连接一个任务，天然支持并发）；async with server保证退出时优雅关闭；get_extra_info('peername')获取对端地址。

3. 获取 HTTP 响应头（含 HTTPS 场景）

import asyncio import urllib.parse import sys async def print_http_headers(url): url = urllib.parse.urlsplit(url) if url.scheme == 'https': reader, writer = await asyncio.open_connection( url.hostname, 443, ssl=True) else: reader, writer = await asyncio.open_connection( url.hostname, 80) query = ( f"HEAD {url.path or '/'} HTTP/1.0\r\n" f"Host: {url.hostname}\r\n" f"\r\n" ) writer.write(query.encode('latin-1')) while True: line = await reader.readline() if not line: break line = line.decode('latin1').rstrip() if line: print(f'HTTP header> {line}') # Ignore the body, close the socket writer.close() url = sys.argv[1] asyncio.run(print_http_headers(url))

用法：

python example.py http://example.com/path/page.html # 或 HTTPS python example.py https://example.com/path/page.html

该示例展示了两个实用技巧：open_connection(..., ssl=True)一行启用 TLS；await reader.readline()逐行读取 HTTP 响应头（HTTP 头正是以\r\n行分隔的文本协议）。

4. 注册已打开的 socket 等待数据

import asyncio import socket async def wait_for_data(): # Get a reference to the current event loop because # we want to access low-level APIs. loop = asyncio.get_running_loop() # Create a pair of connected sockets. rsock, wsock = socket.socketpair() # Register the open socket to wait for data. reader, writer = await asyncio.open_connection(sock=rsock) # Simulate the reception of data from the network loop.call_soon(wsock.send, 'abc'.encode()) # Wait for data data = await reader.read(100) # Got data, we are done: close the socket print("Received:", data.decode()) writer.close() # Close the second socket wsock.close() asyncio.run(wait_for_data())

要点：open_connection(sock=rsock)传入已存在的 socket 而非 host/port，即可把既有 socket 纳入 asyncio 事件循环管理；socket.socketpair()创建本机进程内互连的一对 socket，适合自测。

本文档在当前仓库中的位置与适用前提

本文所讲解的 asyncio Streams 文档位于 asyncio-stream.rst，属于仓库内嵌 Python 运行时（kbe/src/lib/python/）的标准库文档副本。当前内嵌版本为Python 3.7.3（见 patchlevel.h 中的PY_VERSION "3.7.3"），因此：

• 文中标注versionadded:: 3.7的 API（如ssl_handshake_timeout、start_serving、wait_closed、is_closing）在当前运行时中均可使用；
• 更早引入的readuntil（3.5.2）、异步迭代协议等能力同样齐备；
• 相关配套文档可继续参阅 asyncio.rst（asyncio 总览）与 asyncio-api-index.rst（API 索引）；
• 若读者在 KBEngine 的 Python 脚本层（服务端 assets）开发中需要构建 HTTP 探测、内嵌管理接口或工具脚本，这套内嵌的 asyncio 高层 API 可直接使用，无需额外安装依赖。

小结

asyncio Streams 是「用协程思维写网络 IO」的最终形态：四个顶层函数解决连接与服务端的建立，StreamReader的read/readline/readexactly/readuntil覆盖了按字节、按行、精确量与按分隔符四种读取模式，StreamWriter的write/drain/close/wait_closed构成完整的写入与关闭闭环。配合源码可见的 64 KiB 默认缓冲、双倍缓冲暂停读取的背压机制，以及协程回调自动 Task 化调度，这套 API 既能写出教科书式的回声服务器，也能胜任真实的 HTTP/自定义协议客户端。所有示例与文档原文均可在仓库内 asyncio-stream.rst 中找到，实现细节可对照 streams.py 逐一验证。

• 游戏开发
• 后端