python asyncio tcp server_asyncio异步IO——Streams详解
前言
本文翻譯自python3.7官方文檔——asyncio-stream,譯者馬鳴謙,郵箱 1612557569@qq.com。轉載請注明出處。
數據流(Streams)
數據流(Streams)是用于處理網絡連接的高階異步/等待就緒(async/await-ready)原語,可以在不使用回調和底層傳輸協議的情況下發送和接收數據。
以下是一個用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())
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!'))
完整代碼見例子一節。
Stream方法
以下所列的高層asyncio方法可以被用作創建和處理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)
創建一個網絡連接，并返回一對(reader,writer)對象。
返回的reader和writer對象是StreamReader和StreamWriter類的實例。
loop是可選參數，在此方法被某個協程await時能夠自動確定。
limit限定返回的StreamReader實例使用的緩沖區大小。默認情況下，緩沖區限制為64KiB。
其余的參數被直接傳遞給loop.create_connection()。
python3.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)
啟動一個socket服務端。
client_connected_cb指定的回調函數，在新連接建立的時候被調用。該回調函數接收StreamReader和StreamWriter類的‘實例對’(reader,writer)作為兩個參數。
client_connected_cb可以是普通的可調用函數，也可以是協程函數。如果是協程函數，那么會被自動封裝為Task對象處理。
loop是可選參數，在此方法被某個協程await時能夠自動確定。
limit限定返回的StreamReader實例使用的緩沖區大小。默認情況下，緩沖區限定值為64KiB。
其余的參數被直接傳遞給loop.create_server()。
python3.7新增：ssl_handshake_timeout和start_serving參數。
Unix Sockets
coroutine asyncio.open_unix_connection(path=None,*,loop=None,limit=None,ssl=None,sock=None,server_hostname=None,ssl_handshake_timeout=None)
創建一個Unix socket連接，并返回一對(reader,writer)對象。
與open_connection類似，只是運行在Unix sockets上。
可用于：Unix
python3.7新增：ssl_handshake_timeout參數。
python3.7修正：path參數可以為類path(path-like)對象
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 sockets上。
可用于：Unix
python3.7新增：ssl_handshake_timeout參數。
python3.7修正：path參數可以為類path(path-like)對象
StreamReader
class asyncio.StreamReader
定義一個讀取器對象，提供從IO數據流中讀取數據的API。
不建議 直接實例化StreamReader對象。建議通過open_connection()或start_server()創建此類對象。
coroutine read(n=-1)
最多讀取n字節數據。如果n未設置，或被設置為-1，則讀取至EOF標志，并返回讀到的所有字節。
如果在緩沖區仍為空時遇到EOF，則返回一個空的bytes對象。
coroutine readline()
讀取一行(以\n為標志)。
如果在找到\n之前遇到EOF，則返回已讀取到的數據段。
如果遇到EOF時內部緩沖區仍為空，則返回空的bytes對象。
coroutine readexactly(n)
精確讀取n字節數據。
如果在尚未讀夠n字節時遇到EOF，則引發IncompleteReadError異常。已經讀取的部分數據可以通過IncompleteReadError.partial屬性獲取。
coroutine readuntil(separator=b'\n')
從數據流中讀取數據直到遇到separator。
如果執行成功，讀到的數據和分隔符將從內部緩沖區里移除。返回的數據會在末尾包含分隔符。
如果讀取數據的總量超過了配置的數據流緩沖區限制，則引發LimitOverrunError,數據會被留在內部緩沖區中，可以被再次讀取。
如果在找到separator分隔符之前遇到EOF，則引發IncompleteReadError異常，內部緩沖區會被重置。IncompleteReadError.partial屬性會包含部分separator。
python3.5.2新增。
at_eof()
如果緩沖區為空，且feed_eof()被調用，則返回True。
StreamWriter
class asyncio.StreamWriter
定義一個寫入器對象，提供向IO數據流中寫入數據的API。
不建議直接實例化StreamWriter對象，建議通過open_connection或start_server實例化對象。
can_writer_eof()
如果下層傳輸支持write_eof方法，則返回True,否則返回False。
write_eof()
在緩沖的寫入數據被刷新后，關閉數據流的寫入端。
transport
返回下層的asyncio傳輸。
get_extra_info(name,default=None)
訪問可選的傳輸信息。
write(data)
向數據流中寫入數據。
此方法不受流量控制的影響。write()應同drain()一同使用。
writelines()
向數據流中寫入bytes列表(或任何的可迭代對象)。
此方法不受流量控制的影響。應與drain()一同使用。
coroutine drain()
等待恢復數據寫入的時機。例如：
writer.write(data)
await writer.drain()
這是一個與底層IO輸入緩沖區交互的流量控制方法。當緩沖區達到上限時，drain()阻塞，待到緩沖區回落到下限時，寫操作可以被恢復。當不需要等待時，drain()會立即返回。
close()
關閉數據流。
is_closing()
如果數據流已經關閉或正在關閉，則返回True。
coroutine wait_closed()
保持等待，直到數據流關閉。
保持等待，直到底層連接被關閉，應該在close()后調用此方法。
Python3.7新增。
示例
利用Stream實現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實現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()
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注冊等待數據的開放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())
總結
以上是生活随笔為你收集整理的python asyncio tcp server_asyncio异步IO——Streams详解的全部內容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： c++ string类_C++｜细说ST
- 下一篇： activiti 文档_免费、开源、多平