游戏开发后端【免费下载链接】kbengineA MMOG engine of server.项目地址https://gitcode.com/gh_mirrors/kb/kbengine点击查看免费下载本篇技术指南以当前仓库内嵌 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(host, port, ...)建立 TCP 网络连接返回(reader, writer)对start_server(client_connected_cb, host, port, ...)启动 TCP socket 服务器open_unix_connection(path, ...)建立 Unix socket 连接仅 Unix 平台start_unix_server(client_connected_cb, path, ...)启动 Unix socket 服务器仅 Unix 平台open_connection客户端连接入口签名如下ssl_handshake_timeout自 Python 3.7 起可用asyncio.open_connection(hostNone, portNone, *, loopNone, limitNone, sslNone, family0, proto0, flags0, sockNone, local_addrNone, server_hostnameNone, ssl_handshake_timeoutNone)返回的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(limitlimit)再包装为StreamReaderProtocol随后调用loop.create_connection(lambda: protocol, host, port, **kwds)拿到传输层最终组装出StreamWriter并返回(reader, writer)元组。文档注释也坦诚说明如果想自定义StreamReader或StreamReaderProtocol直接复制这段代码即可——它本质上只是便利封装。start_server服务器启动入口asyncio.start_server(client_connected_cb, hostNone, portNone, *, loopNone, limitNone, familysocket.AF_UNSPEC, flagssocket.AI_PASSIVE, sockNone, backlog100, sslNone, reuse_addressNone, reuse_portNone, ssl_handshake_timeoutNone, start_servingTrue)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_serving3.7 新增为False时只创建服务器而不立即开始服务其余参数透传给loop.create_server返回值与loop.create_server一致即Server对象可用于停止服务。在 base_events.py 中可以看到这些参数的底层落实例如reuse_address在 POSIX非 Cygwin平台上默认开启reuse_address os.name posix and sys.platform ! cygwinbacklog被用于sock.listen(backlog)ssl_handshake_timeout则在非 SSL 场景下会被校验拒绝。Unix socket 变体asyncio.open_unix_connection(pathNone, *, loopNone, limitNone, sslNone, sockNone, server_hostnameNone, ssl_handshake_timeoutNone) asyncio.start_unix_server(client_connected_cb, pathNone, *, loopNone, limitNone, sockNone, backlog100, sslNone, ssl_handshake_timeoutNone, start_servingTrue)与 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缓冲区读取操作本质上是「等数据 → 从缓冲区取走」。读取方法方法行为await read(n-1)读取最多n字节n缺省或为-1时一直读到 EOF 并返回全部字节若已收到 EOF 且内部缓冲区为空返回空bytesawait readline()读取一行以\n结尾的字节序列EOF 到达而未找到\n时返回已读到的部分数据await readexactly(n)精确读取n字节EOF 提前到达则抛IncompleteReadError可用其partial属性取得部分数据await readuntil(separatorb\n)读取直到找到separator返回数据包含分隔符超出流限制抛LimitOverrunError数据保留在缓冲区中可再读EOF 前未找到完整分隔符抛IncompleteReadError缓冲区被重置at_eof()缓冲区为空且已调用feed_eof时返回True细节要点read(0)立即返回空字节readexactly(0)返回空字节负数参数抛ValueErrorstreams.py#L623-L625、L664-L665readline内部复用readuntil(b\n)超限时若缓冲区以换行开头则把完整行移除否则清空缓冲区并抛ValueErrorStreamReader还实现了异步迭代协议async for line in reader:等价于逐行readline()空字节时抛出StopAsyncIteration结束迭代streams.py#L690-L697。缓冲、限流与并发约束源码级原理默认限制 64 KiB_DEFAULT_LIMIT 2 ** 16streams.py#L19limit必须大于 0否则构造时抛ValueError背压backpressurefeed_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)属性partialEOF 前已读取的字节与expected期望的总字节数未知时为None由readexactly、readuntil抛出LimitOverrunError在查找分隔符时缓冲区达到限制而抛出属性consumed表示已被消费的字节总数数据仍保留在内部缓冲区中。两者均定义在 streams.py#L22-L50并在 asyncio-exceptions.rst 中被单独归档说明。StreamWriter流式写入 APIStreamWriter是对Transport的薄封装streams.py#L276-L284同样不建议直接实例化。写入与关闭方法方法/属性行为write(data)写入数据不受流控约束调用后应跟await drain()writelines(data)写入字节列表或任意可迭代对象同样需配合drain()await drain()等待合适时机继续写是流控方法当写缓冲区达到高水位时阻塞直到排空到低水位才恢复close()关闭流await wait_closed()等待流完全关闭应在close()之后调用3.7 新增is_closing()流已关闭或正在关闭时返回True3.7 新增write_eof()在缓冲数据刷新后关闭流的写端需底层传输支持can_write_eof()底层传输是否支持write_eof()transport返回底层 asyncio transportget_extra_info(name, defaultNone)访问传输的可选信息如peername对端地址drain 的流控原理drain()是 Streams 与低层协议流控衔接的核心streams.py#L328-L348若关联 reader 已有异常先抛出该异常若传输正在关闭先await sleep(0)让事件循环有机会触发connection_lost()否则循环中write(); await drain()会永不感知连接错误委托_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(fSend: {message!r}) writer.write(message.encode()) data await reader.read(100) print(fReceived: {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(fReceived {message!r} from {addr!r}) print(fSend: {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(fServing 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, sslTrue) else: reader, writer await asyncio.open_connection( url.hostname, 80) query ( fHEAD {url.path or /} HTTP/1.0\r\n fHost: {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(fHTTP 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(..., sslTrue)一行启用 TLSawait 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(sockrsock) # 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(sockrsock)传入已存在的 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在当前运行时中均可使用更早引入的readuntil3.5.2、异步迭代协议等能力同样齐备相关配套文档可继续参阅 asyncio.rstasyncio 总览与 asyncio-api-index.rstAPI 索引若读者在 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 逐一验证。赞分享游戏开发后端【免费下载链接】kbengineA MMOG engine of server.项目地址https://gitcode.com/gh_mirrors/kb/kbengine点击查看免费下载相关推荐KBEngine 内嵌 Python 3.7 的 _thread 低级线程 API 完全指南KBEngine 内嵌 Python 3.7 的 _thread 低级线程 API 完全指南 本文以 KBEngine 仓库内嵌 Python 3.7.3 标准游戏开发后端KBEngine 中的嵌入式 PythonWindows 平台上运行、链接与内嵌 Python 的技术指南KBEngine 中的嵌入式 PythonWindows 平台上运行、链接与内嵌 Python 的技术指南 本篇以 CPython 官方 FAQ 中的《Pyt游戏开发后端KBEngine 内嵌 Python 之 Distutils 实战编写 Setup Script 的完整指南KBEngine 内嵌 Python 之 Distutils 实战编写 Setup Script 的完整指南 导读 本文以 KBEngine 仓库内嵌 Pyt游戏开发后端上一篇Windows驱动清理神器Driver Store Explorer完全指南与实用技巧下一篇Windows驱动存储清理终极指南3个步骤高效释放C盘空间创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考