传输和协议

序言

传输和协议由低级事件循环 API 使用，例如 loop.create_connection()。它们使用基于回调的编程风格，并支持网络或 IPC 协议（例如 HTTP）的高性能实现。

从本质上讲，传输和协议只应在库和框架中使用，而不能在高级 asyncio 应用程序中使用。

此文档页面涵盖 传输 和 协议。

简介

在最高级别，传输关注如何传输字节，而协议确定哪些字节传输（以及在某种程度上何时传输）。

另一种说法：从传输的角度来看，传输是套接字（或类似 I/O 端点）的抽象，而协议是应用程序的抽象。

另一种观点是传输和协议接口共同定义了一个抽象接口，用于使用网络 I/O 和进程间 I/O。

传输和协议对象之间始终存在 1:1 关系：协议调用传输方法发送数据，而传输调用协议方法传递已接收的数据。

大多数面向连接的事件循环方法（例如 loop.create_connection()）通常接受一个protocol_factory 参数，用于为接受的连接（由Transport 对象表示）创建一个Protocol 对象。此类方法通常返回 (transport, protocol) 元组。

内容

此文档页面包含以下部分

传输

源代码： Lib/asyncio/transports.py

传输是 asyncio 提供的类，用于抽象各种通信通道。

传输对象始终由 asyncio 事件循环 实例化。

asyncio 为 TCP、UDP、SSL 和子进程管道实现了传输。传输上可用的方法取决于传输的类型。

传输类不是线程安全的。

传输层次结构

class asyncio.BaseTransport

所有传输的基本类。包含所有 asyncio 传输共享的方法。

类 asyncio.WriteTransport(BaseTransport)

仅写入连接的基本传输。

WriteTransport 类的实例从 loop.connect_write_pipe() 事件循环方法返回，也由子进程相关方法使用，如 loop.subprocess_exec()。

类 asyncio.ReadTransport(BaseTransport)

仅读取连接的基本传输。

ReadTransport 类的实例从 loop.connect_read_pipe() 事件循环方法返回，也由子进程相关方法使用，如 loop.subprocess_exec()。

类 asyncio.Transport(WriteTransport, ReadTransport)

表示双向传输的接口，如 TCP 连接。

用户不直接实例化传输；他们调用一个实用函数，向其传递协议工厂和其他创建传输和协议所需的信息。

Transport 类的实例由事件循环方法返回或使用，如 loop.create_connection()、loop.create_unix_connection()、loop.create_server()、loop.sendfile() 等。

类 asyncio.DatagramTransport(BaseTransport)

数据报 (UDP) 连接的传输。

DatagramTransport 类的实例从 loop.create_datagram_endpoint() 事件循环方法返回。

class asyncio.SubprocessTransport(BaseTransport)

一个抽象，表示父进程与其子操作系统进程之间的连接。

SubprocessTransport 类的实例从事件循环方法 loop.subprocess_shell() 和 loop.subprocess_exec() 返回。

基本传输

BaseTransport.close()

关闭传输。

如果传输具有用于传出数据的缓冲区，则缓冲数据将异步刷新。不会再接收更多数据。在所有缓冲数据刷新后，将使用 None 作为其参数调用协议的 protocol.connection_lost() 方法。一旦关闭传输，就不应再使用它。

BaseTransport.is_closing()

如果传输正在关闭或已关闭，则返回 True。

BaseTransport.get_extra_info(name, default=None)

返回有关传输或其使用的基础资源的信息。

name 是表示要获取的传输特定信息片段的字符串。

default 是如果信息不可用，或者传输不支持使用给定的第三方事件循环实现或在当前平台上查询它，则要返回的值。

例如，以下代码尝试获取传输的基础套接字对象

sock = transport.get_extra_info('socket')
if sock is not None:
    print(sock.getsockopt(...))

可在某些传输上查询的信息类别

BaseTransport.set_protocol(protocol)

设置新协议。

仅当两个协议都有文档支持切换时，才应切换协议。

BaseTransport.get_protocol()

返回当前协议。

只读传输

ReadTransport.is_reading()

如果传输正在接收新数据，则返回 True。

在 3.7 版中添加。

ReadTransport.pause_reading()

暂停传输的接收端。在调用 resume_reading() 之前，不会将任何数据传递给协议的 protocol.data_received() 方法。

在 3.7 版中更改：此方法是幂等的，即在传输已暂停或关闭时可以调用此方法。

ReadTransport.resume_reading()

恢复接收端。如果有一些数据可供读取，则再次调用协议的 protocol.data_received() 方法。

在 3.7 版中更改：此方法是幂等的，即在传输已读取时可以调用此方法。

只写传输

WriteTransport.abort()

立即关闭传输，而不等待挂起的操作完成。缓冲数据将丢失。不会再接收更多数据。最终将使用 None 作为其参数调用协议的 protocol.connection_lost() 方法。

WriteTransport.can_write_eof()

如果传输支持 write_eof()，则返回 True，如果不支持，则返回 False。

WriteTransport.get_write_buffer_size()

返回传输使用的输出缓冲区的当前大小。

WriteTransport.get_write_buffer_limits()

获取写入流控制的高和低水位线。返回一个元组 (low, high)，其中low和high是正字节数。

使用 set_write_buffer_limits() 设置限制。

在 3.4.2 版本中添加。

WriteTransport.set_write_buffer_limits(high=None, low=None)

设置写入流控制的高和低水位线。

这两个值（以字节数衡量）控制何时调用协议的 protocol.pause_writing() 和 protocol.resume_writing() 方法。如果指定，则低水位线必须小于或等于高水位线。high 和 low 都不能为负数。

pause_writing() 在缓冲区大小大于或等于high值时调用。如果已暂停写入，则在缓冲区大小小于或等于low值时调用 resume_writing()。

默认值因实现而异。如果只给出了高水位线，则低水位线默认为小于或等于高水位线的实现特定值。将 high 设置为零也会将 low 强制为零，并导致在缓冲区变为非空时调用 pause_writing()。将 low 设置为零会导致仅在缓冲区为空时调用 resume_writing()。对任一限制使用零通常不是最佳选择，因为它减少了同时进行 I/O 和计算的机会。

使用 get_write_buffer_limits() 获取限制。

WriteTransport.write(data)

将一些 data 字节写入传输。

此方法不会阻塞；它会缓冲数据并安排异步发送数据。

WriteTransport.writelines(list_of_data)

将数据字节列表（或任何可迭代对象）写入传输。这在功能上等效于对可迭代对象产生的每个元素调用 write()，但实现效率可能更高。

WriteTransport.write_eof()

在刷新所有缓冲数据后关闭传输的写入端。仍可以接收数据。

如果传输（例如 SSL）不支持半关闭连接，则此方法可能会引发 NotImplementedError。

数据报传输

DatagramTransport.sendto(data, addr=None)

将 data 字节发送到由 addr（传输相关的目标地址）给出的远程对等方。如果 addr 为 None，则数据将发送到传输创建时给出的目标地址。

此方法不会阻塞；它会缓冲数据并安排异步发送数据。

DatagramTransport.abort()

立即关闭传输，而不等待挂起的操作完成。缓冲数据将丢失。不会再接收更多数据。最终将使用 None 作为其参数调用协议的 protocol.connection_lost() 方法。

子进程传输

SubprocessTransport.get_pid()

以整数形式返回子进程进程 ID。

SubprocessTransport.get_pipe_transport(fd)

返回对应于整数文件描述符 fd 的通信管道传输

  • 0：标准输入（stdin）的可读流传输，或 None（如果子进程不是使用 stdin=PIPE 创建的）

  • 1：标准输出（stdout）的可写流传输，或 None（如果子进程不是使用 stdout=PIPE 创建的）

  • 2：标准错误（stderr）的可写流传输，或 None（如果子进程不是使用 stderr=PIPE 创建的）

  • 其他 fd：None

SubprocessTransport.get_returncode()

返回子进程返回代码（整数形式），或 None（如果子进程尚未返回），类似于 subprocess.Popen.returncode 属性。

SubprocessTransport.kill()

终止子进程。

在 POSIX 系统上，此函数会向子进程发送 SIGKILL 信号。在 Windows 上，此方法是 terminate() 的别名。

另请参阅 subprocess.Popen.kill()。

SubprocessTransport.send_signal(signal)

向子进程发送 signal 编号，如 subprocess.Popen.send_signal() 中所示。

SubprocessTransport.terminate()

停止子进程。

在 POSIX 系统上，此方法会向子进程发送 SIGTERM。在 Windows 上，会调用 Windows API 函数 TerminateProcess() 来停止子进程。

另请参阅 subprocess.Popen.terminate()。

SubprocessTransport.close()

通过调用 kill() 方法来终止子进程。

如果子进程尚未返回，则关闭 stdin、stdout 和 stderr 管道的传输。

协议

源代码： Lib/asyncio/protocols.py

asyncio 提供了一组抽象基类，应使用这些基类来实现网络协议。这些类旨在与 传输 一起使用。

抽象基协议类的子类可以实现部分或全部方法。所有这些方法都是回调：它们在特定事件（例如接收到某些数据）时由传输调用。基协议方法应由相应的传输调用。

基本协议

class asyncio.BaseProtocol

具有所有协议共享的方法的基本协议。

class asyncio.Protocol(BaseProtocol)

实现流式协议（TCP、Unix 套接字等）的基本类。

class asyncio.BufferedProtocol(BaseProtocol)

实现具有手动接收缓冲区控制的流式协议的基本类。

类 asyncio.DatagramProtocol(BaseProtocol)

用于实现数据报（UDP）协议的基类。

类 asyncio.SubprocessProtocol(BaseProtocol)

用于实现与子进程通信的协议（单向管道）的基类。

基本协议

所有 asyncio 协议都可以实现基本协议回调。

连接回调

连接回调在所有协议上调用，每个成功连接一次。所有其他协议回调只能在这两个方法之间调用。

BaseProtocol.connection_made(transport)

在建立连接时调用。

transport 参数是表示连接的传输。协议负责存储对其传输的引用。

BaseProtocol.connection_lost(exc)

在连接丢失或关闭时调用。

参数是异常对象或 None。后者表示收到常规 EOF，或者连接已中止或由连接的此侧关闭。

流控制回调

流控制回调可以由传输调用以暂停或恢复协议执行的写入。

有关更多详细信息，请参阅 set_write_buffer_limits() 方法的文档。

BaseProtocol.pause_writing()

在传输的缓冲区超过高水位线时调用。

BaseProtocol.resume_writing()

在传输的缓冲区低于低水位线时调用。

如果缓冲区大小等于高水位线，则不调用 pause_writing()：缓冲区大小必须严格超过。

相反，当缓冲区大小等于或低于低水位线时，调用 resume_writing()。这些结束条件对于确保在任一标记为零时按预期进行非常重要。

流式协议

事件方法，例如 loop.create_server()、loop.create_unix_server()、loop.create_connection()、loop.create_unix_connection()、loop.connect_accepted_socket()、loop.connect_read_pipe() 和 loop.connect_write_pipe() 接受返回流协议的工厂。

Protocol.data_received(data)

在收到某些数据时调用。data 是包含传入数据的非空字节对象。

数据是缓冲、分块还是重新组装取决于传输。通常，您不应该依赖于特定语义，而应使您的解析通用且灵活。但是，数据始终按正确顺序接收。

在连接打开时，可以任意多次调用该方法。

但是，protocol.eof_received() 最多调用一次。一旦调用 eof_received()，将不再调用 data_received()。

Protocol.eof_received()

当另一端发出信号表示它不会再发送任何数据时调用（例如，如果另一端也使用 asyncio，则调用 transport.write_eof()）。

此方法可以返回假值（包括 None），在这种情况下，传输将自行关闭。相反，如果此方法返回真值，则使用的协议将确定是否关闭传输。由于默认实现返回 None，因此它会隐式关闭连接。

包括 SSL 在内的一些传输不支持半关闭连接，在这种情况下，从此方法返回 true 将导致连接关闭。

状态机

start -> connection_made
    [-> data_received]*
    [-> eof_received]?
-> connection_lost -> end

缓冲流协议

在 3.7 版中添加。

缓冲协议可与支持 流协议 的任何事件循环方法一起使用。

BufferedProtocol 实现允许显式手动分配和控制接收缓冲区。然后，事件循环可以使用协议提供的缓冲区来避免不必要的数据复制。对于接收大量数据的协议，这可以显着提高性能。复杂的协议实现可以显著减少缓冲区分配的数量。

以下回调在 BufferedProtocol 实例上调用

BufferedProtocol.get_buffer(sizehint)

调用以分配新的接收缓冲区。

sizehint 是返回缓冲区的建议最小大小。返回比 sizehint 建议的缓冲区更小或更大是可以接受的。当设置为 -1 时，缓冲区大小可以是任意的。返回大小为零的缓冲区是错误的。

get_buffer() 必须返回一个实现 buffer 协议 的对象。

BufferedProtocol.buffer_updated(nbytes)

在使用接收到的数据更新缓冲区时调用。

nbytes 是写入缓冲区的总字节数。

BufferedProtocol.eof_received()

请参阅 protocol.eof_received() 方法的文档。

get_buffer() 可以在连接期间任意多次调用。但是，protocol.eof_received() 最多调用一次，如果调用，get_buffer() 和 buffer_updated() 在调用后将不会被调用。

状态机

start -> connection_made
    [-> get_buffer
        [-> buffer_updated]?
    ]*
    [-> eof_received]?
-> connection_lost -> end

数据报协议

数据报协议实例应由传递给 loop.create_datagram_endpoint() 方法的协议工厂构造。

DatagramProtocol.datagram_received(data, addr)

在收到数据报时调用。data 是包含传入数据的字节对象。addr 是发送数据的对等方的地址；确切的格式取决于传输方式。

DatagramProtocol.error_received(exc)

当先前的发送或接收操作引发 OSError 时调用。exc 是 OSError 实例。

当传输（例如 UDP）检测到数据报无法传递给其接收者时，在罕见的情况下会调用此方法。但在许多情况下，不可传递的数据报将被静默丢弃。

注意

在 BSD 系统（macOS、FreeBSD 等）上，数据报协议不支持流控制，因为没有可靠的方法来检测因写入过多数据包而导致的发送失败。

套接字始终显示为“就绪”，并且会丢弃多余的数据包。可能会或可能不会引发 OSError，其中 errno 设置为 errno.ENOBUFS；如果引发，它将报告给 DatagramProtocol.error_received()，否则将被忽略。

子进程协议

子进程协议实例应由传递给 loop.subprocess_exec() 和 loop.subprocess_shell() 方法的协议工厂来构建。

SubprocessProtocol.pipe_data_received(fd, data)

当子进程将其数据写入 stdout 或 stderr 管道时调用。

fd 是管道的整数文件描述符。

data 是包含接收数据的非空字节对象。

SubprocessProtocol.pipe_connection_lost(fd, exc)

当与子进程通信的管道之一关闭时调用。

fd 是关闭的整数文件描述符。

SubprocessProtocol.process_exited()

当子进程退出时调用。

可以在 pipe_data_received() 和 pipe_connection_lost() 方法之前调用。

示例

TCP 回显服务器

使用 loop.create_server() 方法创建 TCP 回显服务器，发送回接收到的数据，然后关闭连接

import asyncio


class EchoServerProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        peername = transport.get_extra_info('peername')
        print('Connection from {}'.format(peername))
        self.transport = transport

    def data_received(self, data):
        message = data.decode()
        print('Data received: {!r}'.format(message))

        print('Send: {!r}'.format(message))
        self.transport.write(data)

        print('Close the client socket')
        self.transport.close()


async def main():
    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()

    server = await loop.create_server(
        lambda: EchoServerProtocol(),
        '127.0.0.1', 8888)

    async with server:
        await server.serve_forever()


asyncio.run(main())

另请参阅

使用流的 TCP 回显服务器 示例使用高级 asyncio.start_server() 函数。

TCP 回显客户端

使用 loop.create_connection() 方法的 TCP 回显客户端，发送数据，然后等到连接关闭

import asyncio


class EchoClientProtocol(asyncio.Protocol):
    def __init__(self, message, on_con_lost):
        self.message = message
        self.on_con_lost = on_con_lost

    def connection_made(self, transport):
        transport.write(self.message.encode())
        print('Data sent: {!r}'.format(self.message))

    def data_received(self, data):
        print('Data received: {!r}'.format(data.decode()))

    def connection_lost(self, exc):
        print('The server closed the connection')
        self.on_con_lost.set_result(True)


async def main():
    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()

    on_con_lost = loop.create_future()
    message = 'Hello World!'

    transport, protocol = await loop.create_connection(
        lambda: EchoClientProtocol(message, on_con_lost),
        '127.0.0.1', 8888)

    # Wait until the protocol signals that the connection
    # is lost and close the transport.
    try:
        await on_con_lost
    finally:
        transport.close()


asyncio.run(main())

另请参阅

使用流的 TCP 回显客户端 示例使用高级 asyncio.open_connection() 函数。

UDP 回显服务器

使用 loop.create_datagram_endpoint() 方法的 UDP 回显服务器，发送回接收到的数据

import asyncio


class EchoServerProtocol:
    def connection_made(self, transport):
        self.transport = transport

    def datagram_received(self, data, addr):
        message = data.decode()
        print('Received %r from %s' % (message, addr))
        print('Send %r to %s' % (message, addr))
        self.transport.sendto(data, addr)


async def main():
    print("Starting UDP server")

    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()

    # One protocol instance will be created to serve all
    # client requests.
    transport, protocol = await loop.create_datagram_endpoint(
        lambda: EchoServerProtocol(),
        local_addr=('127.0.0.1', 9999))

    try:
        await asyncio.sleep(3600)  # Serve for 1 hour.
    finally:
        transport.close()


asyncio.run(main())

UDP 回显客户端

使用 loop.create_datagram_endpoint() 方法的 UDP 回显客户端，发送数据，并在收到答复时关闭传输

import asyncio


class EchoClientProtocol:
    def __init__(self, message, on_con_lost):
        self.message = message
        self.on_con_lost = on_con_lost
        self.transport = None

    def connection_made(self, transport):
        self.transport = transport
        print('Send:', self.message)
        self.transport.sendto(self.message.encode())

    def datagram_received(self, data, addr):
        print("Received:", data.decode())

        print("Close the socket")
        self.transport.close()

    def error_received(self, exc):
        print('Error received:', exc)

    def connection_lost(self, exc):
        print("Connection closed")
        self.on_con_lost.set_result(True)


async def main():
    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()

    on_con_lost = loop.create_future()
    message = "Hello World!"

    transport, protocol = await loop.create_datagram_endpoint(
        lambda: EchoClientProtocol(message, on_con_lost),
        remote_addr=('127.0.0.1', 9999))

    try:
        await on_con_lost
    finally:
        transport.close()


asyncio.run(main())

连接现有套接字

使用带有协议的 loop.create_connection() 方法等到套接字接收数据

import asyncio
import socket


class MyProtocol(asyncio.Protocol):

    def __init__(self, on_con_lost):
        self.transport = None
        self.on_con_lost = on_con_lost

    def connection_made(self, transport):
        self.transport = transport

    def data_received(self, data):
        print("Received:", data.decode())

        # We are done: close the transport;
        # connection_lost() will be called automatically.
        self.transport.close()

    def connection_lost(self, exc):
        # The socket has been closed
        self.on_con_lost.set_result(True)


async def main():
    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()
    on_con_lost = loop.create_future()

    # Create a pair of connected sockets
    rsock, wsock = socket.socketpair()

    # Register the socket to wait for data.
    transport, protocol = await loop.create_connection(
        lambda: MyProtocol(on_con_lost), sock=rsock)

    # Simulate the reception of data from the network.
    loop.call_soon(wsock.send, 'abc'.encode())

    try:
        await protocol.on_con_lost
    finally:
        transport.close()
        wsock.close()

asyncio.run(main())

另请参阅

使用低级 loop.add_reader() 方法注册 FD 的 监视文件描述符以了解读取事件 示例。

使用高级流在协程中注册开放套接字以等待数据，该流由 open_connection() 函数创建的 示例。

loop.subprocess_exec() 和 SubprocessProtocol

用于获取子进程的输出并等待子进程退出的子进程协议示例。

子进程由 loop.subprocess_exec() 方法创建

import asyncio
import sys

class DateProtocol(asyncio.SubprocessProtocol):
    def __init__(self, exit_future):
        self.exit_future = exit_future
        self.output = bytearray()
        self.pipe_closed = False
        self.exited = False

    def pipe_connection_lost(self, fd, exc):
        self.pipe_closed = True
        self.check_for_exit()

    def pipe_data_received(self, fd, data):
        self.output.extend(data)

    def process_exited(self):
        self.exited = True
        # process_exited() method can be called before
        # pipe_connection_lost() method: wait until both methods are
        # called.
        self.check_for_exit()

    def check_for_exit(self):
        if self.pipe_closed and self.exited:
            self.exit_future.set_result(True)

async def get_date():
    # Get a reference to the event loop as we plan to use
    # low-level APIs.
    loop = asyncio.get_running_loop()

    code = 'import datetime; print(datetime.datetime.now())'
    exit_future = asyncio.Future(loop=loop)

    # Create the subprocess controlled by DateProtocol;
    # redirect the standard output into a pipe.
    transport, protocol = await loop.subprocess_exec(
        lambda: DateProtocol(exit_future),
        sys.executable, '-c', code,
        stdin=None, stderr=None)

    # Wait for the subprocess exit using the process_exited()
    # method of the protocol.
    await exit_future

    # Close the stdout pipe.
    transport.close()

    # Read the output which was collected by the
    # pipe_data_received() method of the protocol.
    data = bytes(protocol.output)
    return data.decode('ascii').rstrip()

date = asyncio.run(get_date())
print(f"Current date: {date}")

另请参阅使用高级 API 编写的相同示例。