流¶
流是用于处理网络连接的高级 async/await 就绪原语。流允许发送和接收数据,而无需使用回调或低级协议和传输。
以下是用 asyncio 流编写的 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())
await writer.drain()
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 函数可用于创建和使用流
- coroutine asyncio.open_connection(host=None, port=None, *, limit=None, ssl=None, family=0, proto=0, flags=0, sock=None, local_addr=None, server_hostname=None, ssl_handshake_timeout=None, ssl_shutdown_timeout=None, happy_eyeballs_delay=None, interleave=None)¶
建立网络连接并返回一对
(reader, writer)
对象。返回的 reader 和 writer 对象是
StreamReader
和StreamWriter
类的实例。limit 确定返回的
StreamReader
实例使用的缓冲区大小限制。默认情况下,limit 设置为 64 KiB。其余参数直接传递给
loop.create_connection()
。注意
sock 参数将套接字的所有权转移到创建的
StreamWriter
。要关闭套接字,请调用其close()
方法。在版本 3.7 中更改: 添加了 ssl_handshake_timeout 参数。
在版本 3.8 中更改: 添加了 happy_eyeballs_delay 和 interleave 参数。
在版本 3.10 中更改: 删除了 loop 参数。
在版本 3.11 中更改: 添加了 ssl_shutdown_timeout 参数。
- coroutine asyncio.start_server(client_connected_cb, host=None, port=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, ssl_shutdown_timeout=None, start_serving=True)¶
启动一个套接字服务器。
每当建立新的客户端连接时,都会调用 client_connected_cb 回调。它接收一对
(reader, writer)
作为两个参数,它们是StreamReader
和StreamWriter
类的实例。client_connected_cb 可以是普通可调用对象或 协程函数;如果它是协程函数,它将自动作为
Task
调度。limit 确定返回的
StreamReader
实例使用的缓冲区大小限制。默认情况下,limit 设置为 64 KiB。其余参数直接传递给
loop.create_server()
。注意
sock 参数将套接字的所有权转移到创建的服务器。要关闭套接字,请调用服务器的
close()
方法。在版本 3.7 中更改: 添加了 ssl_handshake_timeout 和 start_serving 参数。
在版本 3.10 中更改: 删除了 loop 参数。
在版本 3.11 中更改: 添加了 ssl_shutdown_timeout 参数。
Unix 套接字
- coroutine asyncio.open_unix_connection(path=None, *, limit=None, ssl=None, sock=None, server_hostname=None, ssl_handshake_timeout=None, ssl_shutdown_timeout=None)¶
建立 Unix 套接字连接并返回一对
(reader, writer)
。类似于
open_connection()
,但它在 Unix 套接字上运行。另请参阅
loop.create_unix_connection()
的文档。注意
sock 参数将套接字的所有权转移到创建的
StreamWriter
。要关闭套接字,请调用其close()
方法。可用性: Unix。
3.7 版中的变更: 添加了 ssl_handshake_timeout 参数。path 参数现在可以是 路径类对象
在版本 3.10 中更改: 删除了 loop 参数。
在版本 3.11 中更改: 添加了 ssl_shutdown_timeout 参数。
- 协程 asyncio.start_unix_server(client_connected_cb, path=None, *, limit=None, sock=None, backlog=100, ssl=None, ssl_handshake_timeout=None, ssl_shutdown_timeout=None, start_serving=True)¶
启动 Unix 套接字服务器。
类似于
start_server()
,但适用于 Unix 套接字。另请参阅
loop.create_unix_server()
的文档。注意
sock 参数将套接字的所有权转移到创建的服务器。要关闭套接字,请调用服务器的
close()
方法。可用性: Unix。
3.7 版中的变更: 添加了 ssl_handshake_timeout 和 start_serving 参数。path 参数现在可以是 路径类对象.
在版本 3.10 中更改: 删除了 loop 参数。
在版本 3.11 中更改: 添加了 ssl_shutdown_timeout 参数。
StreamReader¶
- 类 asyncio.StreamReader¶
表示一个读取器对象,提供用于从 IO 流读取数据的 API。作为 异步可迭代对象,该对象支持
async for
语句。不建议直接实例化 StreamReader 对象;请使用
open_connection()
和start_server()
代替。- feed_eof()¶
确认 EOF。
- 协程 read(n=-1)¶
从流中读取最多 n 个字节。
如果未提供 n 或将其设置为
-1
,则读取直到 EOF,然后返回所有已读取的bytes
。如果收到 EOF 并且内部缓冲区为空,则返回一个空的bytes
对象。如果 n 为
0
,则立即返回一个空的bytes
对象。如果 n 为正数,则在内部缓冲区中至少有 1 个字节可用时,立即返回最多 n 个可用的
bytes
。如果在读取任何字节之前收到 EOF,则返回一个空的bytes
对象。
- 协程 readline()¶
读取一行,其中“行”是结尾为
\n
的字节序列。如果收到 EOF 并且未找到
\n
,则该方法将返回部分读取的数据。如果收到 EOF 并且内部缓冲区为空,则返回一个空的
bytes
对象。
- 协程 readexactly(n)¶
精确读取 n 个字节。
如果在可以读取 n 个字节之前遇到 EOF,则引发
IncompleteReadError
。使用IncompleteReadError.partial
属性获取部分读取的数据。
- 协程 readuntil(separator=b'\n')¶
从流中读取数据,直到找到 separator。
成功后,数据和分隔符将从内部缓冲区中删除(消耗)。返回的数据将在末尾包含分隔符。
如果在找到完整分隔符之前遇到 EOF,则会引发
LimitOverrunError
异常,并且数据将保留在内部缓冲区中,可以再次读取。如果在找到完整分隔符之前遇到 EOF,则会引发
IncompleteReadError
异常,并且内部缓冲区将重置。IncompleteReadError.partial
属性可能包含分隔符的一部分。3.5.2 版新增。
- at_eof()¶
如果缓冲区为空并且调用了
feed_eof()
,则返回True
。
StreamWriter¶
- 类 asyncio.StreamWriter¶
表示一个写入器对象,提供用于将数据写入 IO 流的 API。
不建议直接实例化 StreamWriter 对象;请使用
open_connection()
和start_server()
代替。- write(data)¶
该方法尝试立即将 data 写入底层套接字。如果失败,则将数据排队到内部写入缓冲区,直到可以发送为止。
该方法应与
drain()
方法一起使用stream.write(data) await stream.drain()
- writelines(data)¶
该方法立即将字节列表(或任何可迭代对象)写入底层套接字。如果失败,则将数据排队到内部写入缓冲区,直到可以发送为止。
该方法应与
drain()
方法一起使用stream.writelines(lines) await stream.drain()
- close()¶
该方法关闭流和底层套接字。
该方法应与
wait_closed()
方法一起使用,尽管不是必需的。stream.close() await stream.wait_closed()
- can_write_eof()¶
如果底层传输支持
write_eof()
方法,则返回True
,否则返回False
。
- write_eof()¶
在刷新缓冲的写入数据后关闭流的写入端。
- transport¶
返回底层的 asyncio 传输。
- get_extra_info(name, default=None)¶
访问可选的传输信息;有关详细信息,请参阅
BaseTransport.get_extra_info()
。
- 协程 drain()¶
等待直到可以恢复写入流。示例
writer.write(data) await writer.drain()
这是一种与底层 IO 写入缓冲区交互的流量控制方法。当缓冲区的大小达到高水位线时,drain() 将阻塞,直到缓冲区的大小下降到低水位线,并且可以恢复写入。当没有要等待的内容时,
drain()
将立即返回。
- 协程 start_tls(sslcontext, *, server_hostname=None, ssl_handshake_timeout=None, ssl_shutdown_timeout=None)¶
将现有的基于流的连接升级到 TLS。
参数
sslcontext:
SSLContext
的已配置实例。server_hostname: 设置或覆盖目标服务器证书将与之匹配的主机名。
ssl_handshake_timeout 是在 TLS 握手完成之前等待的时间(以秒为单位),超过此时间将中止连接。如果为
None
(默认值),则为60.0
秒。ssl_shutdown_timeout 是在 SSL 关闭完成之前等待的时间(以秒为单位),超过此时间将中止连接。如果为
None
(默认值),则为30.0
秒。
在版本 3.11 中添加。
在版本 3.12 中更改: 添加了 ssl_shutdown_timeout 参数。
- is_closing()¶
如果流已关闭或正在关闭,则返回
True
。在版本 3.7 中添加。
示例¶
使用流的 TCP 回显客户端¶
使用 asyncio.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())
await writer.drain()
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!'))
另请参阅
TCP 回显客户端协议 示例使用低级 loop.create_connection()
方法。
使用流的 TCP 回显服务器¶
使用 asyncio.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()
await writer.wait_closed()
async def main():
server = await asyncio.start_server(
handle_echo, '127.0.0.1', 8888)
addrs = ', '.join(str(sock.getsockname()) for sock in server.sockets)
print(f'Serving on {addrs}')
async with server:
await server.serve_forever()
asyncio.run(main())
另请参阅
TCP 回显服务器协议 示例使用 loop.create_server()
方法。
获取 HTTP 标头¶
查询命令行上传递的 URL 的 HTTP 标头的简单示例
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()
await writer.wait_closed()
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()
函数的协程,等待套接字接收数据
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()
await writer.wait_closed()
# Close the second socket
wsock.close()
asyncio.run(wait_for_data())
另请参阅
使用协议注册一个打开的套接字以等待数据 示例使用低级协议和 loop.create_connection()
方法。
监视文件描述符以获取读取事件 示例使用低级 loop.add_reader()
方法来监视文件描述符。