Python3中的asyncio

blackgoose 发布于3年前
0 条问题

最近想换换口味,在用 asyncio 写一个小东西,过程中碰到各种概念上、实践上的问题,悄悄 记在这里XD.

所谓的“异步”

回归到最初的定义的话,“异步”是指不同硬件之间可以工作在 不同的时钟信号下——试想一下要求所有硬件工作在相同时钟信 号下的系统该有多脆弱。所以同步总线通常出现在与系统本身 工作时钟接近的硬件接口上,而异步总线正好相反,用来连接 远远达不到系统工作频率的硬件。

这些概念投射到同根同源的软件上的话,由于软件命令最终都 是由硬件去执行的,所以软件的操作也有快有慢,“异步操作” 的重点就是协调“快”的操作和“慢”的操作。然后众所周知,最 慢的操作是IO.

与PC硬件里琳琅满目的总线速度不同,软件里一般只有两种速 度:指令执行速度和IO速度——像Erlang计算reduction的时候 也只考虑指令执行时间和IO时间,所有的指令和IO类型都是一 视同仁的。这是为神马呢?大概是因为人脑跟不上电脑吧……

“正确”的异步操作

网上到处都是这样的例子:

 
import socket
import multiprocessing

def handler(conn, addr):
    message = conn.recv(1024)
    while message:
        conn.send(message)
    conn.close()

def server(host, port):
    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    listener.bind((host, port))
    listener.listen(10)

    while True:
        conn, addr = listener.accept()
        process = multiprocessing.Process(target=handler, args=(conn, addr))
        process.start()

这是异步操作没错,用另一个进程来处理请求,只是粒度略大, 因为这只是将快的进程(handler进程)和慢的进程(server进 程)分开了而已,无论在handler还是server里都要等待IO. 而 且我在#python上贴出 这个程序的时候马上有一堆人出来说服务器不应该这样写之类 的XD.

于是武林中就有了下面这种模式:

 
import socket
import selectors

def handler(conn):
    message = conn.recv(1024)
    conn.send(message)

def server(host, port):
    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    listener.bind((host, port))
    listener.listen(10)

    selector = selectors.DefaultSelector()
    selector.register(listener, selectors.EVENT_READ)

    while True:
        event_list = selector.select()
        for key, events in event_list:
            conn = key.fileobj
            if conn == listener:
                new_conn, addr = conn.accept()
                selector.register(new_conn, selectors.EVENT_READ)
            else:
                handler(conn)

我将这个模式称为“在忙完指令之后等待IO”,坊间说的“异步IO” 一般也是特指这种两段式模式:一段等待IO,一段执行指令。别 看上面的程序很简单,很大一部分高性能的服务器或者框架(像 Tornado、Twisted和Erlang)都是基于这个模式的,因为它的资 源消耗比基于多进程/多线程的服务器实在是少太多了(参考Nginx 和Apache的对比),所以扩展性(scalability)也好太多了。

asyncio模块

asyncio是在Python 3.4中添加的新模块,实现了上面的“忙完指 令之后等待IO”模式。

这世上已经有好多异步框架和库了,Guido老爷子为什么要推行这 样一个新模块?他在 PEP 里说的原因是,这些第三方异步代码相互之间不兼容不能移植 blahblah……我倒觉得是Twisted的camelCaseNaming不合老爷子胃 口而已……

还是上面的echo server例子,用外星科技实现:

 
import asyncio

class EchoProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        self.transport = transport

    def connection_lost(self, exc):
        self.transport.close()

    def data_received(self, data):
        self.transport.write(data)

def server(host, port):
    loop = asyncio.get_event_loop()
    srv = loop.create_server(EchoProtocol, host, port)
    asyncio.async(srv)
    loop.run_forever()

熟悉Twisted的同学应该会有deja vu的感觉,我们提供一个接收 事件的对象(实际上是产生对象的factory),然后事件就源源不 断地自动出现了。

但是这看起来和之前selector的例子不像啊?提示:真正的循环 在 BaseEventLoop的_run_once方法里 :)

实践中使用asyncio的要点

好了上面的例子很美好,但其实只展示了asyncio一小部分的威力, 下面来一些私人干货。

在事件处理方法中创建TCP连接

由于实际上asyncio是在事件循环中调用asyncio.Protocol类 (或者子类)的data_received等方法的,这些事件处理方法 如果阻塞的话,会将整个事件循环也阻塞住,失去了所有“异 步IO”模式带来的好处,所以所有的事件处理方法——包括 data_received、connection_made、connection_lost等——都 不能调用任何可能阻塞的函数,包括socket对象的recv方法、 文件对象的read方法等,当然socket的connect方法由于域名 解析和网络延迟等也是会阻塞的……那我们要怎么从事件处理 方法里做connect操作呢?

答案是asyncio提供了一系列异步操作的、不会阻塞的接口, 当然也包括“创建TCP连接”。这些接口全部以coroutine的形 式提供,调用时要使用 yield from 语法。例如可以这样搭 建一个简单的代理服务器:

 
HOST = 'www.google.com'
PORT = '80'

@asyncio.coroutine
def new_connection(host, port):
    loop = asyncio.get_event_loop()
    client_transport, client_proto = yield from \
        loop.create_connection(ClientProtocol, host, port)
    return client_transport, client_proto

class ClientProtocol(asyncio.Protocol):
    ....

class ServerProtocol(asyncio.Protocol):
    def connection_made(self, transport):
        self.transport = transport
        self.client_task = asyncio.Task(new_connection(HOST, PORT))
        self.client_task.add_done_callback(self.client_connect_done)
        ....

    def client_connect_done(self, future):
        client_transport, client_proto = future.result()
        ....

def server(host, port):
    loop = asyncio.get_event_loop()
    srv = loop.create_server(ProxyProtocol, host, port)
    asyncio.async(srv)
    loop.run_forever()

ServerProtocol在收到一个新连接(connection_made)的时候用 asyncio.Task调度一个创建新连接的异步函数,这个函数会由asyncio 的事件循环在connection_made返回后择机执行,执行完成后事 件循环再去调用通过add_done_callback注册的处理函数 (client_connect_done),满满的javascript既视感啊有木有。

为什么yield from不能直接写在connection_made里,而需要另外 封装一个函数呢?因为asyncio的事件循环认定了Protocol对象的 事件处理方法是普通函数,如果yield from直接出现在 connection_made中的话,事件循环调用connection_made的时候 只会返回一个generator,connection_made的函数体完全不会被 执行,所以在事件处理方法中只能通过调度Task(或者使用 asyncio.async(...),效果一样)的方式执行异步操作。

替换事件循环使用的selector

Python 3.4中还有一个和asyncio配套的新模块:selectors. 这个模块将select、epoll、kqueue等等系统级异步IO接口抽象 成“selector”类型,规定了统一的对外接口,于是程序只管使 用selector的接口就行了,不用管底层的实现到底是select还 是epoll.

asyncio中用的就是selector模块, asyncio.selector_events.BaseSelectorEventLoop类的构造 函数有一个selector参数,通常使用默认值就可以了,但是 当然我们也可以把它给换成我们自己的类。比方说如果我们希 望事件循环能支持 ZeroMQ 的socket, 可以把selector的底层实现换成zmq.Poller():

 
class ZmqSelector(selectors._BaseSelectorImpl):
    def __init__(self, poller=None):
        super().__init__()
        if poller is not None:
            self._zmq_poller = poller
        else:
            self._zmq_poller = zmq.Poller()

    def _fileobj_lookup(self, fileobj):
        if isinstance(fileobj, zmq.Socket):
            return fileobj
        else:
            return super()._fileobj_lookup(fileobj)

    def register(self, fileobj, events, data=None):
        key = super().register(fileobj, events, data)
        flags = 0
        if events & selectors.EVENT_READ:
            flags |= zmq.POLLIN
        if events & selectors.EVENT_WRITE:
            flags |= zmq.POLLOUT
        self._zmq_poller.register(fileobj, flags)
        return key

    def unregister(self, fileobj):
        key = super().unregister(fileobj)
        self._zmq_poller.unregister(fileobj)
        return key

    def select(self, timeout=None):
        if timeout is not None:
            poll_timeout = max(0, math.ceil(timeout * 1e3))
        else:
            poll_timeout = None

        select_ready = []
        try:
            zmq_events = self._zmq_poller.poll(poll_timeout)
        except zmq.ZMQError as e:
            if e.errno == errno.EINTR:
                return select_ready
            else:
                raise e

        for sock, ev in zmq_events:
            key = self._key_from_fd(sock)
            if key is not None:
                events = 0
                if ev & zmq.POLLIN:
                    events |= selectors.EVENT_READ
                if ev & zmq.POLLOUT:
                    events |= selectors.EVENT_WRITE
                if ev & zmq.POLLERR:
                    events = selectors.EVENT_READ | selectors.EVENT_WRITE
                select_ready.append((key, events & key.events))

        return select_ready


def install_zmq_event_loop():
    event_loop = asyncio.SelectorEventLoop(ZmqSelector())
    asyncio.set_event_loop(event_loop)

将大文件的读写拆成小块

在事件循环里做任何耗时的操作都是不对的,尤其是IO, 即便是可以随时读写的本地文件,内存里装不下的话还是 会启动硬盘马达让你等个半天。最简单的方法是将大文件 的读写拆分成小块,例如每次只读一页的内容:

 
class AsyncFileWrapper(object):
    DEFAULT_BLOCK_SIZE = 4096

    def __init__(self, loop=None, filename=None,
                fileobj=None, mode='rb'):
        if (filename is None and fileobj is None) or \
                (filename is not None and fileobj is not None):
            raise RuntimeError('Confilicting arguments')

        if filename is not None:
            if 'b' not in mode:
                raise RuntimeError('Only binary mode is supported')
            fileobj = open(filename, mode=mode)
        elif 'b' not in fileobj.mode:
            raise RuntimeError('Only binary mode is supported')
        self.fileobj = fileobj

        if loop is None:
            loop = asyncio.get_event_loop()
        self.loop = loop
        self.rbuffer = bytearray()

    def read_ready(self, future, n, total):
        res = self.fileobj.read1(n)

        if not res:     # EOF
            future.set_result(bytes(self.rbuffer))
            return

        self.rbuffer.extend(res)

        if total > 0:
            more_to_go = total - len(self.rbuffer)
            if more_to_go <= 0:     # enough
                res, self.rbuffer = self.rbuffer[:n], self.rbuffer[n:]
                future.set_result(bytes(res))
            else:
                self.loop.call_soon(self.read_ready, future, more_to_go, total)
        else:   # < 0
            self.loop.call_soon(self.read_ready, future, self.DEFAULT_BLOCK_SIZE, total)

    @asyncio.coroutine
    def read(self, n=-1):
        future = asyncio.Future(loop=self.loop)

        if n == 0:
            future.set_result(b'')
            return future
        elif n < 0:
            self.loop.call_soon(self.read_ready, future, self.DEFAULT_BLOCK_SIZE, n)
        else:
            self.loop.call_soon(self.read_ready, future, n, n)

        return future

    def write(self, data):
        # XXX: big data?
        return self.fileobj.write(data)

    def close(self):
        self.fileobj

上面这个类通过不断使用asyncio事件循环的call_soon方法, 重复执行读取小块文件内容的代码(read_ready方法),使得 循环内的其他代码有更多执行机会,典型的以吞吐量换响应速 度。可以在coroutine中这样使用此类:

 
@asyncio.coroutine
def some_func(...):
    ...
    afile = AsyncFileWrapper(filename='some_file.txt')
    content = yield from afile.read()
    ...

Errata

AsyncFileWrapper

多谢Robber Phex评论指正,AsyncFileWrapper wrap起来的文 件对象其实还是工作在同步状态下的,需要指定O_NONBLOCK.

另外这个类实际工作时会出现调用close方法之后read_ready方 法又被事件循环回调的情况,所以close方法中还要做额外的清 理工作。

以上修改都在 这个gist里 .


需要 登录 后回复方可回复, 如果你还没有账号你可以 注册 一个帐号。