流¶

流是高级的、支持 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 函数可用于创建和使用流

async 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 参数。

async 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, keep_alive=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 参数。

版本 3.13 中的新功能: 添加了 keep_alive 参数。

Unix 套接字

async 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 参数。

async 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, cleanup_socket=True)¶

启动 Unix 套接字服务器。

类似于 start_server()，但适用于 Unix 套接字。

如果 cleanup_socket 为 true，则当服务器关闭时，Unix 套接字将自动从文件系统中删除，除非套接字在服务器创建后已被替换。

另请参阅 loop.create_unix_server() 的文档。

备注

sock 参数将套接字的所有权转移到创建的服务器。要关闭套接字，请调用服务器的 close() 方法。

可用性: Unix。

版本 3.7 中的新功能: 添加了 ssl_handshake_timeout 和 start_serving 参数。path 参数现在可以是路径类对象。

版本 3.10 中已更改: 移除了 loop 参数。

版本 3.11 中的新功能: 添加了 ssl_shutdown_timeout 参数。

版本 3.13 中的新功能: 添加了 cleanup_socket 参数。

StreamReader¶

class asyncio.StreamReader¶

表示一个 reader 对象，它提供从 IO 流读取数据的 API。作为异步可迭代对象，该对象支持 async for 语句。

不建议直接实例化 StreamReader 对象；请改用 open_connection() 和 start_server()。

feed_eof()¶: 确认 EOF。

async read(n=-1)¶

从流中读取最多 n 字节。

如果未提供 n 或设置为 -1，则读取直到 EOF，然后返回所有读取的 bytes。如果收到 EOF 且内部缓冲区为空，则返回一个空的 bytes 对象。

如果 n 为 0，则立即返回一个空的 bytes 对象。

如果 n 为正数，则一旦内部缓冲区中至少有 1 个字节可用，就返回最多 n 个可用的 bytes。如果在读取任何字节之前收到 EOF，则返回一个空的 bytes 对象。

async readline()¶

读取一行，其中“行”是以 \n 结尾的字节序列。

如果收到 EOF 且未找到 \n，则该方法返回部分读取的数据。

如果收到 EOF 且内部缓冲区为空，则返回一个空的 bytes 对象。

async readexactly(n)¶

精确读取 n 字节。

如果在读取 n 之前达到 EOF，则引发 IncompleteReadError。使用 IncompleteReadError.partial 属性获取部分读取的数据。

async readuntil(separator=b'\n')¶

从流中读取数据直到找到 separator。

成功后，数据和分隔符将从内部缓冲区中删除（已消费）。返回的数据将在末尾包含分隔符。

如果读取的数据量超过配置的流限制，则会引发 LimitOverrunError 异常，数据将保留在内部缓冲区中，可以再次读取。

如果在找到完整的分隔符之前达到 EOF，则会引发 IncompleteReadError 异常，并重置内部缓冲区。IncompleteReadError.partial 属性可能包含分隔符的一部分。

separator 也可以是分隔符的元组。在这种情况下，返回值将是可能的最短值，其中任何分隔符作为后缀。对于 LimitOverrunError 的目的，最短的分隔符被认为是匹配的分隔符。

3.5.2 版本新增。

版本 3.13 中的新功能: separator 参数现在可以是分隔符的 tuple。

at_eof()¶: 如果缓冲区为空且已调用 feed_eof()，则返回 True。

StreamWriter¶

class asyncio.StreamWriter¶

表示一个 writer 对象，它提供向 IO 流写入数据的 API。

不建议直接实例化 StreamWriter 对象；请改用 open_connection() 和 start_server()。

write(data)¶

该方法尝试立即将 data 写入底层套接字。如果失败，数据将排队到内部写入缓冲区，直到可以发送。

data 缓冲区应该是一个 bytes、bytearray 或 C 连续的一维 memoryview 对象。

该方法应与 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()。

async drain()¶

等待直到适合恢复向流写入。示例

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

这是一种流控制方法，与底层 IO 写入缓冲区交互。当缓冲区大小达到高水位时，drain() 会阻塞，直到缓冲区大小下降到低水位并可以恢复写入。当没有什么可等待时，drain() 会立即返回。

async 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 版本加入。

async wait_closed()¶

等待直到流关闭。

应在 close() 之后调用，以等待底层连接关闭，确保在例如退出程序之前所有数据都已刷新。

在 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() 方法来监视文件描述符。

流¶

StreamReader¶

StreamWriter¶

示例¶

使用流的 TCP 回显客户端¶

使用流的 TCP 回显服务器¶

获取 HTTP 头¶

注册一个打开的套接字以使用流等待数据¶

目录

上一主题

下一主题

本页