20 · asyncio Streams：接收、分帧与连接收尾

这一次跟随一条 TCP 连接

本章输入是一串 TCP 字节，输出是处理函数写回的字节。主线是“监听 → 接受连接 → 按协议读取 → 写回 → 关闭该连接 → 关闭监听”。第 10 节的 async with 只说明进入和退出可以等待；这次把等待放到真实 I/O 边界上。

固定源码来自 tests/test_server_control.py。测试用一个本地 TCP 服务器检查 Slime 的 HTTP transport；HTTPX 只是它的客户端，reader / writer 来自 Python 标准库 asyncio。我们只裁剪服务器职责，不把 HTTPX、Ray 或其他第三方同名方法当成原生 Streams API。

第一步：start_server 把连接交给 serve

server = await asyncio.start_server(serve, "127.0.0.1", 0, ...) 创建监听并返回 asyncio.Server。0 让系统选择空闲端口，server.sockets[0].getsockname()[1] 读取实际端口。默认在返回前开始接受连接；创建 server 本身不等于把某一条连接处理完。

每次新连接到达，回调得到 StreamReader 和 StreamWriter。如果回调是本章这种 async def serve(reader, writer)，asyncio 会把它安排成 Task。并发发生在多条连接处理任务之间；单个 serve 中的读取仍按书写顺序进行。

客户端的对称入口是 reader, writer = await asyncio.open_connection(host, port)。传入的端口必须来自这次 server 的实际监听，而不是硬编码一个可能被占用的端口。

        server = await asyncio.start_server(serve, "127.0.0.1", 0, backlog=2 * concurrency)
        url = f"http://127.0.0.1:{server.sockets[0].getsockname()[1]}"
        try:
            async with httpx.AsyncClient(transport=http_utils._ShardedHTTPTransport(concurrency)) as client:

                async def request(i):
                    body = str(i).encode()
                    async with client.stream("POST", url, content=body) as response:
                        assert await response.aread() == body

                for _ in range(2):
                    await asyncio.gather(*(request(i) for i in range(concurrency + 3)))

第二步：TCP 没有消息边界，Reader 按协议取字节

Slime 测试先 await reader.readuntil(b"\r\n\r\n") 取出完整 HTTP 头，再解析 Content-Length，最后 await reader.readexactly(length) 取 body。两个 await 分别对应两个协议边界。不能把“一次发送”推断成“一次 read 恰好返回完整消息”。

read(0) 立即给出 b""，因此不要仅凭这个返回值判断断线。readuntil 在 EOF 前未见完整分隔符也抛 IncompleteReadError；其 partial 包含已读取部分。超过 reader 的缓冲限制则可能抛 asyncio.LimitOverrunError，这是“边界过长”，与“对端提前结束”不同；不能靠无限增大缓冲替代协议长度校验。

第三步：提前 EOF 是一个有数据的失败

IncompleteReadError.partial 保存不完整字节，expected 在 readexactly(n) 场景记录期待的长度。实验故意声明长度 5，只发 b"xy" 就关闭，服务端得到 partial == b"xy"、expected == 5。不能把这两个字节当作完整成功结果。

真实测试捕获 (asyncio.IncompleteReadError, ConnectionError) 后进入 finally，适合它的持久连接测试：客户端正常结束连接时，下一轮读头也会遇到 EOF。生产协议是否重试、返回错误还是记录截断，要结合当前处理阶段决定。

        async def serve(reader, writer):
            nonlocal active, peak, accepted
            task = asyncio.current_task()
            tasks.add(task)
            active += 1
            accepted += 1
            peak = max(peak, active)
            try:
                while True:
                    headers = await reader.readuntil(b"\r\n\r\n")
                    length = next(
                        (
                            int(line.split(b":", 1)[1])
                            for line in headers.split(b"\r\n")
                            if line.lower().startswith(b"content-length:")
                        ),
                        0,
                    )
                    body = await reader.readexactly(length)
                    if b"/slow " in headers:
                        await asyncio.sleep(0.1)
                    writer.write(b"HTTP/1.1 200 OK\r\nContent-Length: " + str(len(body)).encode() + b"\r\n\r\n" + body)
                    await writer.drain()
            except (asyncio.IncompleteReadError, ConnectionError):
                pass
            finally:
                active -= 1
                writer.close()
                await writer.wait_closed()
                tasks.discard(task)

第四步：write 入队，drain 配合写入流量控制

writer.write(data) 是同步方法，尝试立即写出或把 bytes 留在底层缓冲；它不保证对端已经收到，也不能写成 await writer.write(...)。await writer.drain() 在缓冲压力高时等待恢复可写，压力低时可立即返回。它不是对端业务确认，也不承诺每次都切换任务。

连接收尾用 writer.close() 发起关闭，再 await writer.wait_closed() 等待本端关闭完成。读写或关闭过程中仍可能抛连接错误，所以资源收尾必须放在 finally。真实 serve 在这里先减少 active 计数，再关闭 writer，最后从 handler 集合移除当前 Task；监听器并不替每个连接做这些业务记账。

第五步：监听器和已接受的连接分开收尾

Slime 测试最后先 server.close() / await server.wait_closed()，再 await asyncio.gather(*tasks, return_exceptions=True) 排空记录下来的处理任务。这需要客户端和 handler 配合结束连接；连接仍打开时 wait_closed() 也可能等待，handler 永远等不到协议输入时 gather 同样会一直等，实际服务需要自己的超时或取消策略。

        finally:
            server.close()
            await server.wait_closed()
            await asyncio.gather(*tasks, return_exceptions=True)
        assert active == 0

可运行实验：完整消息与截断消息

下载 example.py，执行 python3 -W error example.py。仅 Python 标准库，推荐 Python 3.12，最低 3.11；只在 127.0.0.1 上临时监听，不访问外部网络。

教学裁剪保留真实 serve(reader, writer)、start_server、读写和收尾接口。把 HTTP 头换成“十进制长度 + \r\n\r\n”，去掉 HTTPX 客户端、连接池压力测试和第三方 transport；额外一个字节用于展示 read(1)，响应首行展示 readline()。Queue 只把处理结果传回实验主任务，确保输出与 handler 关闭都已完成。

"""Cut down from tests/test_server_control.py: local TCP, no HTTP client."""
import asyncio


async def exercise():
    reports = asyncio.Queue()

    async def serve(reader, writer):
        report = None
        try:
            headers = await reader.readuntil(b'\r\n\r\n')
            length = int(headers.split(b'\r\n', 1)[0])
            body = await reader.readexactly(length)
            suffix = await reader.read(1)
            writer.write(b'echo\n' + body)
            await writer.drain()
            report = ('complete', body, suffix)
        except asyncio.IncompleteReadError as error:
            report = ('incomplete', error.partial, error.expected)
        finally:
            writer.close()
            try:
                await writer.wait_closed()
            finally:
                reports.put_nowait(report)

    server = await asyncio.start_server(serve, '127.0.0.1', 0)
    port = server.sockets[0].getsockname()[1]
    print('server listening', server.is_serving())
    try:
        reader, writer = await asyncio.open_connection('127.0.0.1', port)
        try:
            writer.write(b'3\r\n\r\nabc!')
            await writer.drain()
            print('reply line', (await reader.readline()).decode().strip())
            print('reply body', await reader.read())
        finally:
            writer.close()
            await writer.wait_closed()
        print('handler', await reports.get())

        reader, writer = await asyncio.open_connection('127.0.0.1', port)
        writer.write(b'5\r\n\r\nxy')
        await writer.drain()
        writer.close()
        await writer.wait_closed()
        print('handler', await reports.get())
    finally:
        server.close()
        await server.wait_closed()
    print('server listening', server.is_serving())


if __name__ == '__main__':
    asyncio.run(exercise())

server listening True
reply line echo
reply body b'abc'
handler ('complete', b'abc', b'!')
handler ('incomplete', b'xy', 5)
server listening False

按实际调用追踪：第一个客户端发送完整消息 → server 先读分隔符再读固定 body → 写响应并关闭 → 客户端的 read() 因 EOF 返回 body。第二个客户端提前关闭 → server 在 readexactly(5) 抛错 → finally 关闭连接并上报残留字节。最后主任务关闭 server。

语义核对：Streams 与读取/写入方法、Server 生命周期、IncompleteReadError。

专题首页 · 上一节 · 下一节：21 · 异步能力检查与协议边界 · 源码与教程对照 · 全部用法