在 tornado 中异步无阻塞的执行耗时任务

╰半橙微兮° 2021-11-10 03:42 328阅读 0赞

--------------------

在 [linux][] 上 tornado 是基于 epoll 的事件驱动框架，在[网络][Link 1]事件上是无阻塞的。但是因为 tornado 自身是单线程的，所以如果我们在某一个时刻执行了一个耗时的任务，那么就会阻塞在这里，无法响应其他的任务请求，这个和 tornado 的高性能[服务器][Link 2]称号不符，所以我们要想办法把耗时的任务转换为不阻塞主线程，让耗时的任务不影响对其他请求的响应。

在 [python][] 3.2 上，增加了一个并行库 concurrent.futures，这个库提供了更简单的异步执行函数的方法。

如果是在 2.7 之类的 python 版本上，可以使用 `pip install futures` 来安装这个库。

关于这个库的具体使用，这里就不详细展开了，可以去看官方文档，需要注意的是，前两个例子是示例错误的用法，可能会产生死锁。

下面说说如何在 tornado 中结合使用 futures 库，最好的参考莫过于有文档＋代码。正好， tornado 中解析 ip 使用的 dns 解析服务是[多线程][Link 3]无阻塞的。(`netutils.ThreadedResolver`)

我们来看看它的实现，看看如何应用到我们的[程序][Link 4]中来。

#### tornado 中使用多线程无阻塞来处理 dns 请求 ####

# 删除了注释 
    class ThreadedResolver(ExecutorResolver):
        _threadpool = None
        _threadpool_pid = None
        def initialize(self, io_loop=None, num_threads=10):
            threadpool = ThreadedResolver._create_threadpool(num_threads)
            super(ThreadedResolver, self).initialize(
                io_loop=io_loop, executor=threadpool, close_executor=False)
        @classmethod
        def _create_threadpool(cls, num_threads):
            pid = os.getpid()
            if cls._threadpool_pid != pid:
                # Threads cannot survive after a fork, so if our pid isn't what it
                # was when we created the pool then delete it.
                cls._threadpool = None
            if cls._threadpool is None:
                from concurrent.futures import ThreadPoolExecutor
                cls._threadpool = ThreadPoolExecutor(num_threads)
                cls._threadpool_pid = pid
            return cls._threadpool

`ThreadedResolver` 是 `ExecutorEesolver` 的子类，看看它的是实现。

class ExecutorResolver(Resolver):
        def initialize(self, io_loop=None, executor=None, close_executor=True):
            self.io_loop = io_loop or IOLoop.current()
            if executor is not None:
                self.executor = executor
                self.close_executor = close_executor
            else:
                self.executor = dummy_executor
                self.close_executor = False
        def close(self):
            if self.close_executor:
                self.executor.shutdown()
            self.executor = None
        @run_on_executor
        def resolve(self, host, port, family=socket.AF_UNSPEC):
            addrinfo = socket.getaddrinfo(host, port, family, socket.SOCK_STREAM)
            results = []
            for family, socktype, proto, canonname, address in addrinfo:
                results.append((family, address))
            return results

从 `ExecutorResolver` 的实现可以看出来，它的关键参数是 `ioloop` 和 `executor`，干活的 `resolve` 函数被`@run_on_executor` 修饰，结合起来看 `ThreadedResolver` 的实现，那么这里的 `executor` 就是`from concurrent.futures import ThreadPoolExecutor`

再来看看 `@run_on_executor` 的实现。

`run_on_executor` 的实现在 `concurrent.py` 文件中，它的源码如下：

def run_on_executor(fn):
        @functools.wraps(fn)
        def wrapper(self, *args, **kwargs):
            callback = kwargs.pop("callback", None)
            future = self.executor.submit(fn, self, *args, **kwargs)
            if callback:
                self.io_loop.add_future(future,
                                        lambda future: callback(future.result()))
            return future
        return wrapper

*关于 `functions.wraps()` 的介绍可以参考官方文档 functools — Higher-order functions and operations on callable objects*

简单的说，这里对传递进来的函数进行了封装，并用 `self.executor.submit()` 对[包装][Link 5]的函数进行了执行，并判断是否有回调，如果有，就加入到 ioloop 的 callback 里面。

对比官方的 `concurrent.futures.Executor` 的[接口][Link 6]，里面有个 `submit()` 方法，从头至尾看看`ThreadedResolver` 的实现，就是使用了 `concurrent.futures.ThreadPoolExecutor` 这个 `Executor` 的子类。

所以 tornado 中解析 dns 使用的多线程无阻塞的方法的实质就是使用了 `concurrent.futures` 提供的`ThreadPoolExecutor` 功能。

--------------------

#### 使用多线程无阻塞方法来执行耗时的任务 ####

借鉴 tornado 的使用方法，在我们自己的程序中也使用这种方法来处理耗时的任务。

from tornado.concurrent import run_on_executor
    from concurrent.futures import ThreadPoolExecutor
    class LongTimeTask(tornado.web.RequestHandler):
        executor = ThreadPoolExecutor(10)
        @run_on_executor()
        def get(self, data):
            long_time_task(data)

上面就是一个基本的使用方法，下面展示一个使用 sleep() 来模拟耗时的完整程序。

#!/usr/bin/env python
    #-*-coding:utf-8-*-
    import tornado.ioloop
    import tornado.web
    import tornado.httpserver
    from concurrent.futures import ThreadPoolExecutor
    from tornado.concurrent import run_on_executor
    import time
    class App(tornado.web.Application):
        def __init__(self):
            handlers = [
                (r'/', IndexHandler),
                (r'/sleep/(\d+)', SleepHandler),
            ]
            settings = dict()
            tornado.web.Application.__init__(self, handlers, **settings)
    class BaseHandler(tornado.web.RequestHandler):
        executor = ThreadPoolExecutor(10)
    class IndexHandler(tornado.web.RequestHandler):
        def get(self):
            self.write("Hello, world %s" % time.time())
    class SleepHandler(BaseHandler):
        @run_on_executor
        def get(self, n):
            time.sleep(float(n))
            self._callback()
        def _callback(self):
            self.write("after sleep, now I'm back %s" % time.strftime("%Y-%m-%d %H:%M:%S", time.localtime()))
    if __name__ == "__main__":
        app = App()
        server = tornado.httpserver.HTTPServer(app, xheaders=True)
        server.listen(8888)
        tornado.ioloop.IOLoop.instance().start()

此时先调用 127.0.0.1:8888/sleep/10 不会阻塞 127.0.0.1:8888/ 了。

以上，就是完整的在 tornado 中利用多线程来执行耗时的任务。

--------------------

#### 结语 ####

epoll 的好处确实很多，事件就绪通知后，上层任务函数执行任务，如果任务本身需要较耗时，那么就可以考虑这个方法了，  
当然也有其他的方法，比如使用 celery 来调度执行耗时太多的任务，比如频繁的需要写入数据到不同的文件中，我公司的一个[项目][Link 7]中，需要把数据写入四千多个文件中，每天产生几亿条数据，就是使用了 tornado + redis + celery 的方法来高效的执行写文件任务。

完。

原文地址：在 tornado 中异步无阻塞的执行耗时任务, 感谢原作者分享。关键词：

转载于:https://www.cnblogs.com/DjangoBlog/p/5267006.html

[linux]: http://www.07net01.com/linux/
[Link 1]: http://www.wredian.com/tags-%E7%BD%91%E7%BB%9C-0.html
[Link 2]: http://www.07net01.com/tags-%E6%9C%8D%E5%8A%A1%E5%99%A8-0.html
[python]: http://www.07net01.com/tags-python-0.html
[Link 3]: http://www.07net01.com/tags-%E5%A4%9A%E7%BA%BF%E7%A8%8B-0.html
[Link 4]: http://www.07net01.com/tags-%E7%A8%8B%E5%BA%8F-0.html
[Link 5]: http://www.chuyixuexi.com/tags-%E5%8C%85%E8%A3%85-0.html
[Link 6]: http://www.07net01.com/tags-%E6%8E%A5%E5%8F%A3-0.html
[Link 7]: http://www.07net01.com/tags-%E9%A1%B9%E7%9B%AE-0.html