Python传输和协议

传输层级¶

class asyncio.BaseTransport¶

所有传输的基类。包含所有异步IO传输共用的方法。

class asyncio.WriteTransport(BaseTransport)¶

只写链接的基础传输。

Instances of the WriteTransport class are returned from

the loop.connect_write_pipe() event loop method and

are also used by subprocess-related methods like

loop.subprocess_exec().

class asyncio.ReadTransport(BaseTransport)¶

只读链接的基础传输。

Instances of the ReadTransport class are returned from

the loop.connect_read_pipe() event loop method and

are also used by subprocess-related methods like

loop.subprocess_exec().

class asyncio.Transport(WriteTransport, ReadTransport)¶

接口代表一个双向传输，如TCP链接。

用户不用直接实例化传输；调用一个功能函数，给它传递协议工厂和其它需要的信息就可以创建传输和协议。

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

class 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 是在信息不可用或传输不支持第三方事件循环实现或当前平台查询时返回的值。

例如下面代码尝试获取传输相关套接字对象:

python3 notranslate">

sock=transport.get_extra_info('socket')
ifsockisnotNone:
print(sock.getsockopt(...))

传输可查询信息类别:

• 套接字:

  • 'peername': 套接字链接时的远端地址，socket.socket.getpeername() 方法的结果 (出错时为 None )

  • 'socket': socket.socket 实例

  • 'sockname': 套接字本地地址， socket.socket.getsockname() 方法的结果

• SSL套接字

  • 'compression': 用字符串指定压缩算法，或者链接没有压缩时为 None ；ssl.SSLSocket.compression() 的结果。

  • 'cipher': 一个三值的元组，包含使用密码的名称，定义使用的SSL协议的版本，使用的加密位数。 ssl.SSLSocket.cipher() 的结果。

  • 'peercert': 远端凭证； ssl.SSLSocket.getpeercert() 结果。

  • 'sslcontext': ssl.SSLContext 实例

  • 'ssl_object': ssl.SSLObject 或 ssl.SSLSocket 实例

• 管道：

  • 'pipe': 管道对象

• 子进程：

  • 'subprocess': subprocess.Popen 实例

BaseTransport.set_protocol(protocol)¶

设置一个新协议。

只有两种协议都写明支持切换才能完成切换协议。

BaseTransport.get_protocol()¶

返回当前协议。

只读传输¶

ReadTransport.is_reading()¶

如果传输接收到新数据时返回 True 。

3.7 新版功能.

ReadTransport.pause_reading()¶

暂停传输的接收端。protocol.data_received() 方法将不会收到数据，除非 resume_reading()  被调用。

在 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()¶

获取写入流控制 high 和 low 高低标记位。返回元组 (low,high) ， low 和 high 为正字节数。

使用 set_write_buffer_limits() 设置限制。

3.4.2 新版功能.

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

设置写入流控制 high 和 low 高低标记位。

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

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

默认值是实现专属的。 如果只给出了高水位值，则低水位值默认为一个小于或等于高水位值的实现传属值。 将 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()¶

立即关闭传输，不会等待已提交的操作执行完毕。 已缓存的数据将会丢失。 不会接收更多的数据。 协议的 protocol.connection_lost() 方法最终将附带 None 作为参数被调用。

子进程传输¶

SubprocessTransport.get_pid()¶

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

SubprocessTransport.get_pipe_transport(fd)¶

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

• 0: 标准输入 (stdin) 的可读流式传输，如果子进程创建时未设置 stdin=PIPE 则为 None。

• 1: 标准输出writable streaming transport of the standard output (stdout) 的可写流式传输，如果子进程创建时未设置 stdout=PIPE 则为 None。

• 2: 标准错误 (stderr) 的可写流式传输，如果子进程创建时未设置 stderr=PIPE 则为 None

• 其他 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 管道的传输。

基础协议¶

class asyncio.BaseProtocol¶

带有所有协议的共享方法的基础协议。

class asyncio.Protocol(BaseProtocol)¶

用于实现流式协议（TCP, Unix 套接字等等）的基类。

class asyncio.BufferedProtocol(BaseProtocol)¶

用于实现可对接收缓冲区进行手动控制的流式协议的基类。

class asyncio.DatagramProtocol(BaseProtocol)¶

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

class 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()¶

当发出信号的另一端不再继续发送数据时（例如通过调用 transport.write_eof()，如果另一端也使用 asyncio 的话）被调用。

此方法可能返回假值 (包括 None)，在此情况下传输将会自行关闭。 相反地，如果此方法返回真值，将以所用的协议来确定是否要关闭传输。 由于默认实现是返回 None，因此它会隐式地关闭连接。

某些传输，包括 SSL 在内，并不支持半关闭的连接，在此情况下从该方法返回真值将导致连接被关闭。

状态机：

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

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

importasyncio
classEchoServerProtocol(asyncio.Protocol):
defconnection_made(self,transport):
peername=transport.get_extra_info('peername')
print('Connection from {}'.format(peername))
self.transport=transport
defdata_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()
asyncdefmain():
# Get a reference to the event loop as we plan to use
# low-level APIs.
loop=asyncio.get_running_loop()
server=awaitloop.create_server(
lambda:EchoServerProtocol(),
'127.0.0.1',8888)
asyncwithserver:
awaitserver.serve_forever()
asyncio.run(main())

importasyncio
classEchoClientProtocol(asyncio.Protocol):
def__init__(self,message,on_con_lost):
self.message=message
self.on_con_lost=on_con_lost
defconnection_made(self,transport):
transport.write(self.message.encode())
print('Data sent: {!r}'.format(self.message))
defdata_received(self,data):
print('Data received: {!r}'.format(data.decode()))
defconnection_lost(self,exc):
print('The server closed the connection')
self.on_con_lost.set_result(True)
asyncdefmain():
# 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=awaitloop.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:
awaiton_con_lost
finally:
transport.close()
asyncio.run(main())

importasyncio
classEchoServerProtocol:
defconnection_made(self,transport):
self.transport=transport
defdatagram_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)
asyncdefmain():
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=awaitloop.create_datagram_endpoint(
lambda:EchoServerProtocol(),
local_addr=('127.0.0.1',9999))
try:
awaitasyncio.sleep(3600)# Serve for 1 hour.
finally:
transport.close()
asyncio.run(main())

importasyncio
classEchoClientProtocol:
def__init__(self,message,on_con_lost):
self.message=message
self.on_con_lost=on_con_lost
self.transport=None
defconnection_made(self,transport):
self.transport=transport
print('Send:',self.message)
self.transport.sendto(self.message.encode())
defdatagram_received(self,data,addr):
print("Received:",data.decode())
print("Close the socket")
self.transport.close()
deferror_received(self,exc):
print('Error received:',exc)
defconnection_lost(self,exc):
print("Connection closed")
self.on_con_lost.set_result(True)
asyncdefmain():
# 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=awaitloop.create_datagram_endpoint(
lambda:EchoClientProtocol(message,on_con_lost),
remote_addr=('127.0.0.1',9999))
try:
awaiton_con_lost
finally:
transport.close()
asyncio.run(main())

importasyncio
importsocket
classMyProtocol(asyncio.Protocol):
def__init__(self,on_con_lost):
self.transport=None
self.on_con_lost=on_con_lost
defconnection_made(self,transport):
self.transport=transport
defdata_received(self,data):
print("Received:",data.decode())
# We are done: close the transport;
# connection_lost() will be called automatically.
self.transport.close()
defconnection_lost(self,exc):
# The socket has been closed
self.on_con_lost.set_result(True)
asyncdefmain():
# 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=awaitloop.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:
awaitprotocol.on_con_lost
finally:
transport.close()
wsock.close()
asyncio.run(main())

importasyncio
importsys
classDateProtocol(asyncio.SubprocessProtocol):
def__init__(self,exit_future):
self.exit_future=exit_future
self.output=bytearray()
defpipe_data_received(self,fd,data):
self.output.extend(data)
defprocess_exited(self):
self.exit_future.set_result(True)
asyncdefget_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=awaitloop.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.
awaitexit_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)
returndata.decode('ascii').rstrip()
ifsys.platform=="win32":
asyncio.set_event_loop_policy(
asyncio.WindowsProactorEventLoopPolicy())
date=asyncio.run(get_date())
print(f"Current date: {date}")