流?

StreamReader?

class asyncio.StreamReader?

這個(gè)類表示一個(gè)讀取器對(duì)象，該對(duì)象提供api以便于從IO流中讀取數(shù)據(jù)。

不推薦直接實(shí)例化 StreamReader 對(duì)象，建議使用 open_connection() 和 start_server() 來(lái)獲取 StreamReader 實(shí)例。

coroutine read(n=- 1)?

至多讀取 n 個(gè)byte。 如果沒(méi)有設(shè)置 n , 則自動(dòng)置為 -1 ， -1時(shí)表示讀至 EOF 并返回所有讀取的byte。

如果讀到EOF，且內(nèi)部緩沖區(qū)為空，則返回一個(gè)空的 bytes 對(duì)象。

coroutine readline()?

讀取一行，其中“行”指的是以 \n 結(jié)尾的字節(jié)序列。

如果讀到EOF而沒(méi)有找到 \n ，該方法返回部分讀取的數(shù)據(jù)。

如果讀到EOF，且內(nèi)部緩沖區(qū)為空，則返回一個(gè)空的 bytes 對(duì)象。

coroutine readexactly(n)?

精確讀取 n 個(gè) bytes，不會(huì)超過(guò)也不能少于。

如果在讀取完 n 個(gè)byte之前讀取到EOF，則會(huì)引發(fā) IncompleteReadError 異常。使用 IncompleteReadError.partial 屬性來(lái)獲取到達(dá)流結(jié)束之前讀取的 bytes 字符串。

coroutine readuntil(separator=b'\n')?

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

成功后，數(shù)據(jù)和指定的separator將從內(nèi)部緩沖區(qū)中刪除(或者說(shuō)被消費(fèi)掉)。返回的數(shù)據(jù)將包括在末尾的指定separator。

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

如果在找到完整的separator之前到達(dá)EOF，則會(huì)引發(fā) IncompleteReadError 異常，并重置內(nèi)部緩沖區(qū)。 IncompleteReadError.partial 屬性可能包含指定separator的一部分。

3.5.2 新版功能.

at_eof()?

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

StreamWriter?

class asyncio.StreamWriter?

這個(gè)類表示一個(gè)寫(xiě)入器對(duì)象，該對(duì)象提供api以便于寫(xiě)數(shù)據(jù)至IO流中。

不建議直接實(shí)例化 StreamWriter；而應(yīng)改用 open_connection() 和 start_server()。

write(data)?

此方法會(huì)嘗試立即將 data 寫(xiě)入到下層的套接字。 如果寫(xiě)入失敗，數(shù)據(jù)會(huì)被排入內(nèi)部寫(xiě)緩沖隊(duì)列直到可以被發(fā)送。

此方法應(yīng)當(dāng)與 drain() 方法一起使用:

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

writelines(data)?

此方法會(huì)立即嘗試將一個(gè)字節(jié)串列表（或任何可迭代對(duì)象）寫(xiě)入到下層的套接字。 如果寫(xiě)入失敗，數(shù)據(jù)會(huì)被排入內(nèi)部寫(xiě)緩沖隊(duì)列直到可以被發(fā)送。

此方法應(yīng)當(dāng)與 drain() 方法一起使用:

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

close()?

此方法會(huì)關(guān)閉流以及下層的套接字。

此方法應(yīng)與 wait_closed() 方法一起使用:

stream.close()
await stream.wait_closed()

can_write_eof()?

如果下層的傳輸支持 write_eof() 方法則返回``True``，否則返回 False。

write_eof()?

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

transport?

返回下層的 asyncio 傳輸。

get_extra_info(name, default=None)?

訪問(wèn)可選的傳輸信息；詳情參見(jiàn) BaseTransport.get_extra_info()。

coroutine drain()?

等待直到可以適當(dāng)?shù)鼗謴?fù)寫(xiě)入到流。 示例:

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

這是一個(gè)與下層的 IO 寫(xiě)緩沖區(qū)進(jìn)行交互的流程控制方法。 當(dāng)緩沖區(qū)大小達(dá)到最高水位（最大上限）時(shí)，drain() 會(huì)阻塞直到緩沖區(qū)大小減少至最低水位以便恢復(fù)寫(xiě)入。 當(dāng)沒(méi)有要等待的數(shù)據(jù)時(shí)，drain() 會(huì)立即返回。

coroutine start_tls(sslcontext, *, server_hostname=None, ssl_handshake_timeout=None)?

Upgrade an existing stream-based connection to TLS.

Parameters:

• sslcontext: a configured instance of SSLContext.

• server_hostname: sets or overrides the host name that the target server's certificate will be matched against.

• ssl_handshake_timeout is the time in seconds to wait for the TLS handshake to complete before aborting the connection. 60.0 seconds if None (default).

3.11 新版功能.

is_closing()?

如果流已被關(guān)閉或正在被關(guān)閉則返回 True。

3.7 新版功能.

coroutine wait_closed()?

等待直到流被關(guān)閉。

應(yīng)當(dāng)在 close() 之后被調(diào)用以便等待直到下層的連接被關(guān)閉。

3.7 新版功能.

例子?

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

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

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)

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

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

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