前言

本文翻譯自python3.7官方文檔——asyncio-stream,譯者馬鳴謙,郵箱 1612557569@qq.com。轉(zhuǎn)載請(qǐng)注明出處。

數(shù)據(jù)流(Streams)

數(shù)據(jù)流(Streams)是用于處理網(wǎng)絡(luò)連接的高階異步/等待就緒(async/await-ready)原語(yǔ),可以在不使用回調(diào)和底層傳輸協(xié)議的情況下發(fā)送和接收數(shù)據(jù)。

以下是一個(gè)用asyncio實(shí)現(xiàn)的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())

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!'))

完整代碼見例子一節(jié)。

Stream方法

以下所列的高層asyncio方法可以被用作創(chuàng)建和處理Stream：

coroutine asyncio.open_connection(host=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)

創(chuàng)建一個(gè)網(wǎng)絡(luò)連接，并返回一對(duì)(reader,writer)對(duì)象。

返回的reader和writer對(duì)象是StreamReader和StreamWriter類的實(shí)例。

loop是可選參數(shù)，在此方法被某個(gè)協(xié)程await時(shí)能夠自動(dòng)確定。

limit限定返回的StreamReader實(shí)例使用的緩沖區(qū)大小。默認(rèn)情況下，緩沖區(qū)限制為64KiB。

其余的參數(shù)被直接傳遞給loop.create_connection()。

python3.7新增：ssl_handshake_timeout參數(shù)。

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)

啟動(dòng)一個(gè)socket服務(wù)端。

client_connected_cb指定的回調(diào)函數(shù)，在新連接建立的時(shí)候被調(diào)用。該回調(diào)函數(shù)接收StreamReader和StreamWriter類的‘實(shí)例對(duì)’(reader,writer)作為兩個(gè)參數(shù)。

client_connected_cb可以是普通的可調(diào)用函數(shù)，也可以是協(xié)程函數(shù)。如果是協(xié)程函數(shù)，那么會(huì)被自動(dòng)封裝為Task對(duì)象處理。

loop是可選參數(shù)，在此方法被某個(gè)協(xié)程await時(shí)能夠自動(dòng)確定。

limit限定返回的StreamReader實(shí)例使用的緩沖區(qū)大小。默認(rèn)情況下，緩沖區(qū)限定值為64KiB。

其余的參數(shù)被直接傳遞給loop.create_server()。

python3.7新增：ssl_handshake_timeout和start_serving參數(shù)。

Unix Sockets

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

創(chuàng)建一個(gè)Unix socket連接，并返回一對(duì)(reader,writer)對(duì)象。

與open_connection類似，只是運(yùn)行在Unix sockets上。

可用于：Unix

python3.7新增：ssl_handshake_timeout參數(shù)。

python3.7修正：path參數(shù)可以為類path(path-like)對(duì)象

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)

啟動(dòng)一個(gè)Unix socket 服務(wù)端。

類似于start_server,只是運(yùn)行在Unix sockets上。

可用于：Unix

python3.7新增：ssl_handshake_timeout參數(shù)。

python3.7修正：path參數(shù)可以為類path(path-like)對(duì)象

StreamReader

class asyncio.StreamReader

定義一個(gè)讀取器對(duì)象，提供從IO數(shù)據(jù)流中讀取數(shù)據(jù)的API。

不建議 直接實(shí)例化StreamReader對(duì)象。建議通過(guò)open_connection()或start_server()創(chuàng)建此類對(duì)象。

coroutine read(n=-1)

最多讀取n字節(jié)數(shù)據(jù)。如果n未設(shè)置，或被設(shè)置為-1，則讀取至EOF標(biāo)志，并返回讀到的所有字節(jié)。

如果在緩沖區(qū)仍為空時(shí)遇到EOF，則返回一個(gè)空的bytes對(duì)象。

coroutine readline()

讀取一行(以\n為標(biāo)志)。

如果在找到\n之前遇到EOF，則返回已讀取到的數(shù)據(jù)段。

如果遇到EOF時(shí)內(nèi)部緩沖區(qū)仍為空，則返回空的bytes對(duì)象。

coroutine readexactly(n)

精確讀取n字節(jié)數(shù)據(jù)。

如果在尚未讀夠n字節(jié)時(shí)遇到EOF，則引發(fā)IncompleteReadError異常。已經(jīng)讀取的部分?jǐn)?shù)據(jù)可以通過(guò)IncompleteReadError.partial屬性獲取。

coroutine readuntil(separator=b'\n')

從數(shù)據(jù)流中讀取數(shù)據(jù)直到遇到separator。

如果執(zhí)行成功，讀到的數(shù)據(jù)和分隔符將從內(nèi)部緩沖區(qū)里移除。返回的數(shù)據(jù)會(huì)在末尾包含分隔符。

如果讀取數(shù)據(jù)的總量超過(guò)了配置的數(shù)據(jù)流緩沖區(qū)限制，則引發(fā)LimitOverrunError,數(shù)據(jù)會(huì)被留在內(nèi)部緩沖區(qū)中，可以被再次讀取。

如果在找到separator分隔符之前遇到EOF，則引發(fā)IncompleteReadError異常，內(nèi)部緩沖區(qū)會(huì)被重置。IncompleteReadError.partial屬性會(huì)包含部分separator。

python3.5.2新增。

at_eof()

如果緩沖區(qū)為空，且feed_eof()被調(diào)用，則返回True。

StreamWriter

class asyncio.StreamWriter

定義一個(gè)寫入器對(duì)象，提供向IO數(shù)據(jù)流中寫入數(shù)據(jù)的API。

不建議直接實(shí)例化StreamWriter對(duì)象，建議通過(guò)open_connection或start_server實(shí)例化對(duì)象。

can_writer_eof()

如果下層傳輸支持write_eof方法，則返回True,否則返回False。

write_eof()

在緩沖的寫入數(shù)據(jù)被刷新后，關(guān)閉數(shù)據(jù)流的寫入端。

transport

返回下層的asyncio傳輸。

get_extra_info(name,default=None)

訪問(wèn)可選的傳輸信息。

write(data)

向數(shù)據(jù)流中寫入數(shù)據(jù)。

此方法不受流量控制的影響。write()應(yīng)同drain()一同使用。

writelines()

向數(shù)據(jù)流中寫入bytes列表(或任何的可迭代對(duì)象)。

此方法不受流量控制的影響。應(yīng)與drain()一同使用。

coroutine drain()

等待恢復(fù)數(shù)據(jù)寫入的時(shí)機(jī)。例如：

writer.write(data)

await writer.drain()

這是一個(gè)與底層IO輸入緩沖區(qū)交互的流量控制方法。當(dāng)緩沖區(qū)達(dá)到上限時(shí)，drain()阻塞，待到緩沖區(qū)回落到下限時(shí)，寫操作可以被恢復(fù)。當(dāng)不需要等待時(shí)，drain()會(huì)立即返回。

close()

關(guān)閉數(shù)據(jù)流。

is_closing()

如果數(shù)據(jù)流已經(jīng)關(guān)閉或正在關(guān)閉，則返回True。

coroutine wait_closed()

保持等待，直到數(shù)據(jù)流關(guān)閉。

保持等待，直到底層連接被關(guān)閉，應(yīng)該在close()后調(diào)用此方法。

Python3.7新增。

示例

利用Stream實(shí)現(xiàn)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())

data = await reader.read(100)

print(f'Received: {data.decode()!r}')

print('Close the connection')

writer.close()

asyncio.run(tcp_echo_client('Hello World!'))

利用Stream實(shí)現(xiàn)TCP回顯服務(wù)端

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

async def main():

server = await asyncio.start_server(

handle_echo, '127.0.0.1', 8888)

addr = server.sockets[0].getsockname()

print(f'Serving on {addr}')

async with server:

await server.serve_forever()

asyncio.run(main())

獲取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()

url = sys.argv[1]

asyncio.run(print_http_headers(url))

用法：

python example.py http://example.com/path/page.html

或：

python example.py https://example.com/path/page.html

利用Stream注冊(cè)等待數(shù)據(jù)的開放socket

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

# Close the second socket

wsock.close()

asyncio.run(wait_for_data())

總結(jié)

以上是生活随笔為你收集整理的python asyncio tcp server_asyncio异步IO——Streams详解的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。