18.5.5. Streams（基于协程的API）¶

18.5.5.1. Stream的函数¶

None

本模块中的顶层函数仅用作方便包装；有真正没有什么特别的，如果他们不做你想要的，随时复制他们的代码。

coroutine asyncio.open_connection(host=None, port=None, *, loop=None, limit=None, **kwds)¶

create_connection()的包装器返回（rader，writer）对。

reader返回的是一个StreamReader实例；writer是一个StreamWriter实例。

这些参数是AbstractEventLoop.create_connection()的所有常见参数，除了protocol_factory；最常见的是位置主机和端口，各种可选的关键字参数如下。

其他可选的关键字参数是loop（设置事件循环实例使用）和limit（设置传递给StreamReader的缓冲区限制）。

此函数是coroutine。

coroutine asyncio.start_server(client_connected_cb, host=None, port=None, *, loop=None, limit=None, **kwds)¶

启动一个socket服务器，为每个连接的客户端产生一个回调。返回值与create_server()相同。

使用两个参数调用client_connected_cb参数：client_reader，client_writer。client_reader是StreamReader对象，而client_writer是StreamWriter对象。client_connected_cb参数可以是普通回调函数或协程函数；如果它是协程功能，它将被自动转换为任务。

其余的参数是create_server()的所有常见参数，除了protocol_factory；最常见的是位置主机和端口，各种可选的关键字参数如下。

其他可选的关键字参数是loop（设置事件循环实例使用）和limit（设置传递给StreamReader的缓冲区限制）。

此函数是协程。

coroutine asyncio.open_unix_connection(path=None, *, loop=None, limit=None, **kwds)¶

create_unix_connection()的包装器返回（读取器，写入器）对。

有关返回值和其他详细信息的信息，请参见open_connection()。

此函数是coroutine。

None

coroutine asyncio.start_unix_server(client_connected_cb, path=None, *, loop=None, limit=None, **kwds)¶

启动UNIX Domain Socket服务器，为每个连接的客户端回调一个。

有关返回值和其他详细信息的信息，请参见start_server()。

此函数是coroutine。

可用性：UNIX。

18.5.5.2. StreamReader ¶

class asyncio.StreamReader(limit=None, loop=None)¶

此类为不是线程安全的。

exception()¶: None

feed_eof()¶: 确认EOF。

feed_data(data)¶: 在内部缓冲区中输入数据字节。将恢复等待数据的任何操作。

set_exception(exc)¶: None

set_transport(transport)¶: 设置传输。

coroutine read(n=-1)¶

读取最多n个字节。如果未提供n或设置为-1，请读取直到EOF并返回所有读取的字节。

如果接收到EOF并且内部缓冲区为空，则返回一个空的bytes对象。

此方法是coroutine。

coroutine readline()¶

读取一行，其中“line”是以\n结尾的字节序列。

如果接收到EOF，并且未找到\n，则该方法将返回部分读取字节。

如果接收到EOF并且内部缓冲区为空，则返回一个空的bytes对象。

此方法是coroutine。

coroutine readexactly(n)¶

读取n字节。如果读取n之前到达流的末尾，则引发IncompleteReadError，异常的IncompleteReadError.partial属性包含部分读字节。

此方法是协程。

coroutine readuntil(separator=b'n')¶

从流中读取数据，直到找到separator。

成功时，数据和分隔符将从内部缓冲区（消耗）中删除。返回的数据将包括末尾的分隔符。

配置流限制用于检查结果。Limit设置可以返回的数据的最大长度，不计算分隔符。

如果发生EOF并且仍未找到完整的分隔符，则会引发IncompleteReadError异常，并且内部缓冲区将被重置。IncompleteReadError.partial属性可能包含部分分隔符。

如果由于超限而无法读取数据，则会引发LimitOverrunError异常，并且数据将保留在内部缓冲区中，因此可以再次读取。

版本3.5.2中的新功能。

at_eof()¶: 如果缓冲区为空且调用feed_eof()，则返回True。

18.5.5.3. StreamWriter¶

class asyncio.StreamWriter(transport, protocol, reader, loop)¶

包装运输。

这会使write()，writelines()，can_write_eof()，write_eof()，get_extra_info()和close()。它添加drain()，它返回可选的Future，您可以等待流控制。它还添加了直接引用Transport的传输属性。

此类为not thread safe。

transport¶: 运输。

can_write_eof()¶: 如果传输支持write_eof()，False，则返回True请参见WriteTransport.can_write_eof()。

close()¶: 关闭传输：请参阅BaseTransport.close()。

coroutine drain()¶

让底层传输的写缓冲区有机会被刷新。

目的用途是写：

w.write(data)
yield from w.drain()

当传输缓冲区的大小达到高水限值（协议暂停）时，阻塞直到缓冲区的大小下降到低水限值，并且协议恢复。当没有什么要等待，收益率继续立即。

从drain()得到的结果为循环提供了调度写入操作和刷新缓冲区的机会。尤其应当在可能大量的数据写入传输时使用，而协程不会从write()的调用之间产生。

此方法是coroutine。

get_extra_info(name, default=None)¶: 返回可选的传输信息：请参阅BaseTransport.get_extra_info()。

write(data)¶: 将一些数据字节写入传输：参见WriteTransport.write()。

writelines(data)¶: 将数据字节的列表（或任何可迭代的）写入传输：参见WriteTransport.writelines()。

write_eof()¶: 刷新缓冲数据后关闭传输的写入结束：请参阅WriteTransport.write_eof()。

18.5.5.4. StreamReaderProtocol¶

class asyncio.StreamReaderProtocol(stream_reader, client_connected_cb=None, loop=None)¶

Trivial助手类适用于Protocol和StreamReader之间。Protocol的子类。

stream_reader is a StreamReader instance, client_connected_cb is an optional function called with (stream_reader, stream_writer) when a connection is made, loop is the event loop instance to use.

（这是一个帮助类，而不是使StreamReader本身为Protocol子类，因为StreamReader有其他潜在用途， StreamReader意外调用了协议的不当方法。）

18.5.5.5. IncompleteReadError¶

exception asyncio.IncompleteReadError¶

Incomplete read error, subclass of EOFError.

expected¶: 预期字节总数（int）。

partial¶: 在达到流结束之前读取字节字符串（bytes）。

18.5.5.6. LimitOverrunError¶

exception asyncio.LimitOverrunError¶

在查找分隔符时达到缓冲区限制。

consumed¶: 要消耗的总字节数。

18.5.5.7. Stream examples¶

18.5.5.7.1. TCP echo client using streams¶

TCP回显客户端使用asyncio.open_connection()函数：

import asyncio

@asyncio.coroutine
def tcp_echo_client(message, loop):
    reader, writer = yield from asyncio.open_connection('127.0.0.1', 8888,
                                                        loop=loop)

    print('Send: %r' % message)
    writer.write(message.encode())

    data = yield from reader.read(100)
    print('Received: %r' % data.decode())

    print('Close the socket')
    writer.close()

message = 'Hello World!'
loop = asyncio.get_event_loop()
loop.run_until_complete(tcp_echo_client(message, loop))
loop.close()

也可以看看

TCP echo client protocol示例使用AbstractEventLoop.create_connection()方法。

18.5.5.7.2. TCP echo server using streams¶

TCP回显服务器使用asyncio.start_server()函数：

import asyncio

@asyncio.coroutine
def handle_echo(reader, writer):
    data = yield from reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')
    print("Received %r from %r" % (message, addr))

    print("Send: %r" % message)
    writer.write(data)
    yield from writer.drain()

    print("Close the client socket")
    writer.close()

loop = asyncio.get_event_loop()
coro = asyncio.start_server(handle_echo, '127.0.0.1', 8888, loop=loop)
server = loop.run_until_complete(coro)

# Serve requests until Ctrl+C is pressed
print('Serving on {}'.format(server.sockets[0].getsockname()))
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

# Close the server
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()

也可以看看

TCP echo server protocol示例使用AbstractEventLoop.create_server()方法。

18.5.5.7.3. 获取HTTP头¶

在命令行中获取URL的HTTP头的简单示例：

import asyncio
import urllib.parse
import sys

@asyncio.coroutine
def print_http_headers(url):
    url = urllib.parse.urlsplit(url)
    if url.scheme == 'https':
        connect = asyncio.open_connection(url.hostname, 443, ssl=True)
    else:
        connect = asyncio.open_connection(url.hostname, 80)
    reader, writer = yield from connect
    query = ('HEAD {path} HTTP/1.0\r\n'
             'Host: {hostname}\r\n'
             '\r\n').format(path=url.path or '/', hostname=url.hostname)
    writer.write(query.encode('latin-1'))
    while True:
        line = yield from reader.readline()
        if not line:
            break
        line = line.decode('latin1').rstrip()
        if line:
            print('HTTP header> %s' % line)

    # Ignore the body, close the socket
    writer.close()

url = sys.argv[1]
loop = asyncio.get_event_loop()
task = asyncio.ensure_future(print_http_headers(url))
loop.run_until_complete(task)
loop.close()

用法：

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

或使用HTTPS：

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

18.5.5.7.4. Register an open socket to wait for data using streams¶

协程等待，直到套接字使用open_connection()函数接收数据：

import asyncio
try:
    from socket import socketpair
except ImportError:
    from asyncio.windows_utils import socketpair

@asyncio.coroutine
def wait_for_data(loop):
    # Create a pair of connected sockets
    rsock, wsock = socketpair()

    # Register the open socket to wait for data
    reader, writer = yield from asyncio.open_connection(sock=rsock, loop=loop)

    # Simulate the reception of data from the network
    loop.call_soon(wsock.send, 'abc'.encode())

    # Wait for data
    data = yield from reader.read(100)

    # Got data, we are done: close the socket
    print("Received:", data.decode())
    writer.close()

    # Close the second socket
    wsock.close()

loop = asyncio.get_event_loop()
loop.run_until_complete(wait_for_data(loop))
loop.close()

也可以看看

register an open socket to wait for data using a protocol示例使用由AbstractEventLoop.create_connection()方法创建的低级协议注册开放套接字以等待数据。

watch a file descriptor for read events示例使用低级AbstractEventLoop.add_reader()方法注册套接字的文件描述器。

18.5.5. Streams（基于协程的API）¶

18.5.5.1. Stream的函数¶

18.5.5.2. StreamReader ¶

18.5.5.3. StreamWriter¶

18.5.5.4. StreamReaderProtocol¶

18.5.5.5. IncompleteReadError¶

18.5.5.6. LimitOverrunError¶

18.5.5.7. Stream examples¶

18.5.5.7.1. TCP echo client using streams¶

18.5.5.7.2. TCP echo server using streams¶

18.5.5.7.3. 获取HTTP头¶

18.5.5.7.4. Register an open socket to wait for data using streams¶

目录

上一主题

下一主题

这一页