流¶

流是高階的、支援 async/await 的原語，用於處理網路連線。流允許傳送和接收資料，而無需使用回撥或低階協議和傳輸。

這是一個使用 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())
    await writer.drain()

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

另請參閱下面的示例部分。

流函式

以下頂層 asyncio 函式可用於建立和使用流

async asyncio.open_connection(host=None, port=None, *, limit=None, ssl=None, family=0, proto=0, flags=0, sock=None, local_addr=None, server_hostname=None, ssl_handshake_timeout=None, ssl_shutdown_timeout=None, happy_eyeballs_delay=None, interleave=None)¶

建立網路連線並返回一對 (reader, writer) 物件。

返回的 reader 和 writer 物件是 StreamReader 和 StreamWriter 類的例項。

limit 決定了返回的 StreamReader 例項使用的緩衝區大小限制。預設情況下，limit 設定為 64 KiB。

其餘引數直接傳遞給 loop.create_connection()。

備註

sock 引數將套接字的所有權轉移到建立的 StreamWriter。要關閉套接字，請呼叫其 close() 方法。

版本 3.7 中的新功能: 添加了 ssl_handshake_timeout 引數。

版本 3.8 中的新功能: 添加了 happy_eyeballs_delay 和 interleave 引數。

版本 3.10 中已更改: 移除了 loop 引數。

版本 3.11 中的新功能: 添加了 ssl_shutdown_timeout 引數。

async asyncio.start_server(client_connected_cb, host=None, port=None, *, limit=None, family=socket.AF_UNSPEC, flags=socket.AI_PASSIVE, sock=None, backlog=100, ssl=None, reuse_address=None, reuse_port=None, keep_alive=None, ssl_handshake_timeout=None, ssl_shutdown_timeout=None, start_serving=True)¶

啟動套接字伺服器。

每次建立新的客戶端連線時，都會呼叫 client_connected_cb 回撥。它接收一個 (reader, writer) 對作為兩個引數，它們是 StreamReader 和 StreamWriter 類的例項。

client_connected_cb 可以是一個普通的可呼叫物件，也可以是一個協程函式；如果它是一個協程函式，它將自動作為 Task 進行排程。

limit 決定了返回的 StreamReader 例項使用的緩衝區大小限制。預設情況下，limit 設定為 64 KiB。

其餘引數直接傳遞給 loop.create_server()。

備註

sock 引數將套接字的所有權轉移到建立的伺服器。要關閉套接字，請呼叫伺服器的 close() 方法。

版本 3.7 中的新功能: 添加了 ssl_handshake_timeout 和 start_serving 引數。

版本 3.10 中已更改: 移除了 loop 引數。

版本 3.11 中的新功能: 添加了 ssl_shutdown_timeout 引數。

版本 3.13 中的新功能: 添加了 keep_alive 引數。

Unix 套接字

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

建立 Unix 套接字連線並返回一對 (reader, writer)。

類似於 open_connection()，但作用於 Unix 套接字。

另請參閱 loop.create_unix_connection() 的文件。

備註

sock 引數將套接字的所有權轉移到建立的 StreamWriter。要關閉套接字，請呼叫其 close() 方法。

可用性: Unix。

版本 3.7 中的新功能: 添加了 ssl_handshake_timeout 引數。path 引數現在可以是路徑類物件

版本 3.10 中已更改: 移除了 loop 引數。

版本 3.11 中的新功能: 添加了 ssl_shutdown_timeout 引數。

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

啟動 Unix 套接字伺服器。

類似於 start_server()，但適用於 Unix 套接字。

如果 cleanup_socket 為 true，則當伺服器關閉時，Unix 套接字將自動從檔案系統中刪除，除非套接字在伺服器建立後已被替換。

另請參閱 loop.create_unix_server() 的文件。

備註

sock 引數將套接字的所有權轉移到建立的伺服器。要關閉套接字，請呼叫伺服器的 close() 方法。

可用性: Unix。

版本 3.7 中的新功能: 添加了 ssl_handshake_timeout 和 start_serving 引數。path 引數現在可以是路徑類物件。

版本 3.10 中已更改: 移除了 loop 引數。

版本 3.11 中的新功能: 添加了 ssl_shutdown_timeout 引數。

版本 3.13 中的新功能: 添加了 cleanup_socket 引數。

StreamReader¶

class asyncio.StreamReader¶

表示一個 reader 物件，它提供從 IO 流讀取資料的 API。作為非同步可迭代物件，該物件支援 async for 語句。

不建議直接例項化 StreamReader 物件；請改用 open_connection() 和 start_server()。

feed_eof()¶: 確認 EOF。

async read(n=-1)¶

從流中讀取最多 n 位元組。

如果未提供 n 或設定為 -1，則讀取直到 EOF，然後返回所有讀取的 bytes。如果收到 EOF 且內部緩衝區為空，則返回一個空的 bytes 物件。

如果 n 為 0，則立即返回一個空的 bytes 物件。

如果 n 為正數，則一旦內部緩衝區中至少有 1 個位元組可用，就返回最多 n 個可用的 bytes。如果在讀取任何位元組之前收到 EOF，則返回一個空的 bytes 物件。

async readline()¶

讀取一行，其中“行”是以 \n 結尾的位元組序列。

如果收到 EOF 且未找到 \n，則該方法返回部分讀取的資料。

如果收到 EOF 且內部緩衝區為空，則返回一個空的 bytes 物件。

async readexactly(n)¶

精確讀取 n 位元組。

如果在讀取 n 之前達到 EOF，則引發 IncompleteReadError。使用 IncompleteReadError.partial 屬性獲取部分讀取的資料。

async readuntil(separator=b'\n')¶

從流中讀取資料直到找到 separator。

成功後，資料和分隔符將從內部緩衝區中刪除（已消費）。返回的資料將在末尾包含分隔符。

如果讀取的資料量超過配置的流限制，則會引發 LimitOverrunError 異常，資料將保留在內部緩衝區中，可以再次讀取。

如果在找到完整的分隔符之前達到 EOF，則會引發 IncompleteReadError 異常，並重置內部緩衝區。IncompleteReadError.partial 屬性可能包含分隔符的一部分。

separator 也可以是分隔符的元組。在這種情況下，返回值將是可能的最短值，其中任何分隔符作為字尾。對於 LimitOverrunError 的目的，最短的分隔符被認為是匹配的分隔符。

3.5.2 版本新增。

版本 3.13 中的新功能: separator 引數現在可以是分隔符的 tuple。

at_eof()¶: 如果緩衝區為空且已呼叫 feed_eof()，則返回 True。

StreamWriter¶

class asyncio.StreamWriter¶

表示一個 writer 物件，它提供向 IO 流寫入資料的 API。

不建議直接例項化 StreamWriter 物件；請改用 open_connection() 和 start_server()。

write(data)¶

該方法嘗試立即將 data 寫入底層套接字。如果失敗，資料將排隊到內部寫入緩衝區，直到可以傳送。

data 緩衝區應該是一個 bytes、bytearray 或 C 連續的一維 memoryview 物件。

該方法應與 drain() 方法一起使用

stream.write(data)
await stream.drain()

writelines(data)¶

該方法將位元組列表（或任何可迭代物件）立即寫入底層套接字。如果失敗，資料將排隊到內部寫入緩衝區，直到可以傳送。

該方法應與 drain() 方法一起使用

stream.writelines(lines)
await stream.drain()

close()¶

該方法關閉流和底層套接字。

該方法應與 wait_closed() 方法一起使用，儘管不是強制性的

stream.close()
await stream.wait_closed()

can_write_eof()¶: 如果底層傳輸支援 write_eof() 方法，則返回 True，否則返回 False。

write_eof()¶: 重新整理緩衝的寫入資料後，關閉流的寫入端。

transport¶: 返回底層的 asyncio 傳輸。

get_extra_info(name, default=None)¶: 訪問可選的傳輸資訊；有關詳細資訊，請參閱 BaseTransport.get_extra_info()。

async drain()¶

等待直到適合恢復向流寫入。示例

writer.write(data)
await writer.drain()

這是一種流控制方法，與底層 IO 寫入緩衝區互動。當緩衝區大小達到高水位時，drain() 會阻塞，直到緩衝區大小下降到低水位並可以恢復寫入。當沒有什麼可等待時，drain() 會立即返回。

async start_tls(sslcontext, *, server_hostname=None, ssl_handshake_timeout=None, ssl_shutdown_timeout=None)¶

將現有基於流的連線升級到 TLS。

引數

sslcontext: SSLContext 的已配置例項。
server_hostname: 設定或覆蓋目標伺服器證書將匹配的主機名。
ssl_handshake_timeout 是等待 TLS 握手完成的時間（秒），超過此時間將中止連線。None（預設）時為 60.0 秒。
ssl_shutdown_timeout 是等待 SSL 關閉完成的時間（秒），超過此時間將中止連線。None（預設）時為 30.0 秒。

在 3.11 版本中新增。

版本 3.12 中的新功能: 添加了 ssl_shutdown_timeout 引數。

is_closing()¶: 如果流已關閉或正在關閉過程中，則返回 True。

在 3.7 版本加入。

async wait_closed()¶

等待直到流關閉。

應在 close() 之後呼叫，以等待底層連線關閉，確保在例如退出程式之前所有資料都已重新整理。

在 3.7 版本加入。

示例¶

使用流的 TCP 回顯客戶端¶

使用 asyncio.open_connection() 函式的 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())
    await writer.drain()

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

參見

TCP 回顯客戶端協議示例使用低階 loop.create_connection() 方法。

使用流的 TCP 回顯伺服器¶

使用 asyncio.start_server() 函式的 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()
    await writer.wait_closed()

async def main():
    server = await asyncio.start_server(
        handle_echo, '127.0.0.1', 8888)

    addrs = ', '.join(str(sock.getsockname()) for sock in server.sockets)
    print(f'Serving on {addrs}')

    async with server:
        await server.serve_forever()

asyncio.run(main())

參見

TCP 回顯伺服器協議示例使用 loop.create_server() 方法。

獲取 HTTP 頭¶

查詢命令列傳入 URL 的 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()
    await writer.wait_closed()

url = sys.argv[1]
asyncio.run(print_http_headers(url))

用法

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

或使用 HTTPS

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

註冊一個開啟的套接字以使用流等待資料¶

使用 open_connection() 函式等待套接字接收資料的協程

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()
    await writer.wait_closed()

    # Close the second socket
    wsock.close()

asyncio.run(wait_for_data())

參見

註冊一個開啟的套接字以使用協議等待資料示例使用低階協議和 loop.create_connection() 方法。

監視檔案描述符以進行讀取事件示例使用低階 loop.add_reader() 方法來監視檔案描述符。

流¶

StreamReader¶

StreamWriter¶

示例¶

使用流的 TCP 回顯客戶端¶

使用流的 TCP 回顯伺服器¶

獲取 HTTP 頭¶

註冊一個開啟的套接字以使用流等待資料¶

目錄

上一主題

下一主題

本頁