Python流

• Z时代
• 2024-01-10
• 分类：综合

流是用于处理网络连接的高级 async/await-ready 原语。流允许发送和接收数据，而不需要使用回调或低级协议和传输。

下面是一个使用 asyncio streams 编写的 TCP echo 客户端示例:

importasyncio
asyncdeftcp_echo_client(message):
reader,writer=awaitasyncio.open_connection(
'127.0.0.1',8888)
print(f'Send: {message!r}')
writer.write(message.encode())
data=awaitreader.read(100)
print(f'Received: {data.decode()!r}')
print('Close the connection')
writer.close()
awaitwriter.wait_closed()
asyncio.run(tcp_echo_client('Hello World!'))

参见下面的 Examples 部分。

Stream 函数

下面的高级 asyncio 函数可以用来创建和处理流:

coroutine asyncio.open_connection(host=None, port=None, *, loop=None, limit=None, ssl=None, family=0, proto=0, flags=0, sock=None, local_addr=None, server_hostname=None, ssl_handshake_timeout=None)¶

建立网络连接并返回一对 (reader,writer) 对象。

返回的 reader 和 writer 对象是 StreamReader 和 StreamWriter 类的实例。

loop 参数是可选的，当从协程中等待该函数时，总是可以自动确定。

limit 确定返回的 StreamReader 实例使用的缓冲区大小限制。默认情况下，limit 设置为 64 KiB 。

其余的参数直接传递到 loop.create_connection() 。

3.7 新版功能: ssl_handshake_timeout 形参。

coroutine asyncio.start_server(client_connected_cb, host=None, port=None, *, loop=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, start_serving=True)¶

启动套接字服务。

当一个新的客户端连接被建立时，回调函数 client_connected_cb 会被调用。该函数会接收到一对参数 (reader,writer) ，reader是类 StreamReader 的实例，而writer是类 StreamWriter 的实例。

client_connected_cb 即可以是普通的可调用对象也可以是一个 协程函数; 如果它是一个协程函数，它将自动作为 Task 被调度。

loop 参数是可选的。当在一个协程中await该方法时，该参数始终可以自动确定。

limit 确定返回的 StreamReader 实例使用的缓冲区大小限制。默认情况下，limit 设置为 64 KiB 。

余下的参数将会直接传递给 loop.create_server().

3.7 新版功能: The ssl_handshake_timeout and start_serving parameters.

Unix 套接字

coroutine asyncio.open_unix_connection(path=None, *, loop=None, limit=None, ssl=None, sock=None, server_hostname=None, ssl_handshake_timeout=None)¶

建立一个 Unix 套接字连接并返回 (reader,writer) 这对返回值。

与 open_connection() 相似，但是操作在 Unix 套接字上

请看文档 loop.create_unix_connection().

可用性: Unix。

3.7 新版功能: ssl_handshake_timeout 形参。

在 3.7 版更改: path 现在是一个 path-like object

coroutine asyncio.start_unix_server(client_connected_cb, path=None, *, loop=None, limit=None, sock=None, backlog=100, ssl=None, ssl_handshake_timeout=None, start_serving=True)¶

启动一个Unix socket服务。

与 start_server() 相似，但是是在 Unix 套接字上的操作。

请看文档 loop.create_unix_server().

可用性: Unix。

3.7 新版功能: The ssl_handshake_timeout and start_serving parameters.

在 3.7 版更改: path 形参现在可以是 path-like object 对象。

importasyncio
asyncdeftcp_echo_client(message):
reader,writer=awaitasyncio.open_connection(
'127.0.0.1',8888)
print(f'Send: {message!r}')
writer.write(message.encode())
data=awaitreader.read(100)
print(f'Received: {data.decode()!r}')
print('Close the connection')
writer.close()
asyncio.run(tcp_echo_client('Hello World!'))

importasyncio
asyncdefhandle_echo(reader,writer):
data=awaitreader.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)
awaitwriter.drain()
print("Close the connection")
writer.close()
asyncdefmain():
server=awaitasyncio.start_server(
handle_echo,'127.0.0.1',8888)
addr=server.sockets[0].getsockname()
print(f'Serving on {addr}')
asyncwithserver:
awaitserver.serve_forever()
asyncio.run(main())

importasyncio
importurllib.parse
importsys
asyncdefprint_http_headers(url):
url=urllib.parse.urlsplit(url)
ifurl.scheme=='https':
reader,writer=awaitasyncio.open_connection(
url.hostname,443,ssl=True)
else:
reader,writer=awaitasyncio.open_connection(
url.hostname,80)
query=(
f"HEAD {url.pathor'/'} HTTP/1.0\r\n"
f"Host: {url.hostname}\r\n"
f"\r\n"
)
writer.write(query.encode('latin-1'))
whileTrue:
line=awaitreader.readline()
ifnotline:
break
line=line.decode('latin1').rstrip()
ifline:
print(f'HTTP header> {line}')
# Ignore the body, close the socket
writer.close()
url=sys.argv[1]
asyncio.run(print_http_headers(url))

pythonexample.pyhttp://example.com/path/page.html

pythonexample.pyhttps://example.com/path/page.html

importasyncio
importsocket
asyncdefwait_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=awaitasyncio.open_connection(sock=rsock)
# Simulate the reception of data from the network
loop.call_soon(wsock.send,'abc'.encode())
# Wait for data
data=awaitreader.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())