转载

Python3中的asyncio

最近想换换口味，在用 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里 .

原文 http://blog.theerrorlog.com/asyncio-in-python-3.html

正文到此结束

所属分类：编程技术

本文标签： apache 线程域名软件 python java 科技 TCP list http tar 时间 key ask 代码 Select client ip Google 解析 Connection tab 多线程 HTML 同步服务器 UI App 参数 Nginx 进程 CTO 快的
版权声明： 本文为互联网转载文章，出处已在文章中说明(部分除外)。如果侵权，请联系本站长删除，谢谢。
本文海报： 生成海报一生成海报二

其他链接

关于本站

本站定位：个人技术类博客

本站作用：写博客、记日志、闲聊扯淡鼓捣技术。

问题交流

Python3中的asyncio

所谓的“异步”

“正确”的异步操作

asyncio模块

实践中使用asyncio的要点

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

替换事件循环使用的selector

将大文件的读写拆成小块

Errata

AsyncFileWrapper

热门推荐

相关文章

说给你听

本文目录

随机标签

书籍教程

近期评论

网站信息

其他链接

关于本站

问题交流