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



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

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

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

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

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

我们来看看它的实现,看看如何应用到我们的程序中来。

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() 对包装的函数进行了执行,并判断是否有回调,如果有,就加入到 ioloop 的 callback 里面。

对比官方的 concurrent.futures.Executor 的接口,里面有个 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 来调度执行耗时太多的任务,比如频繁的需要写入数据到不同的文件中,我公司的一个项目中,需要把数据写入四千多个文件中,每天产生几亿条数据,就是使用了 tornado + redis + celery 的方法来高效的执行写文件任务。

完。

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

时间: 2024-10-14 04:11:16

在 tornado 中异步无阻塞的执行耗时任务的相关文章

Tornado的异步非阻塞

阻塞和非阻塞Web框架 只有Tornado和Node.js是异步非阻塞的,其他所有的web框架都是阻塞式的. Tornado阻塞和非阻塞两种模式都支持. 阻塞式: 代表:Django.Flask.Tornado.Bottle 一个请求到来未处理完成,后续请求则一直等待. 解决方案:多线程或多进程. 异步非阻塞(存在IO请求): 代表:Tornado(默认单进程/单线程) Tornado的阻塞模式示例 from tornado import ioloop from tornado.web impo

Tornado中异步框架的使用

tornado的同步框架与其他web框架相同都是处理先来的请求,如果先来的请求阻塞,那么后面的请求也会处理不了.一直处于等待过程中.但是请求一旦得到响应,那么: 请求发送过来后,将需要的本站资源直接返回给客户端 请求发送过来后,本站没有需要的资源,从其它站点获取过来,再返回给客户端 一.Tornado中的同步框架 1.本站资源直接返回 import tornado.web import time class LoginHandler(tornado.web.RequestHandler): de

异步无阻塞委托 学习1

using System;using System.Collections.Generic;using System.Linq;using System.Runtime.Remoting.Messaging;using System.Threading;using System.Web;using System.Web.UI;using System.Web.UI.WebControls; namespace WebApplication21{ public partial class WebF

web性能优化之---JavaScript中的无阻塞加载性能优化方案

一.js阻塞特性 JS 有个很无语的阻塞特性,就是当浏览器在执行JS 代码时,不能同时做其他任何事情,无论其代码是内嵌的还是外部的. 即<script>每次出现都会让页面等待脚本的解析和执行(不论JS是内嵌的还是外链的),JS代码执行完成后,才继续渲染页面. 二.优化方案 1.尽管脚本的下载过程并不会相互影响,但页面仍然必须等待所有JS下载并执行完成才能继续.所以尽可能将所有<script>标签放置在页面底部,紧靠关闭标签</body>的上方.此方法可以保证页面在脚本运

Tornado异步非阻塞的使用以及原理

Tornado 和现在的主流 Web 服务器框架(包括大多数 Python 的框架)有着明显的区别:它是非阻塞式服务器,而且速度相当快.得利于其 非阻塞的方式和对 epoll 的运用,Tornado 每秒可以处理数以千计的连接,这意味着对于实时 Web 服务来说,Tornado 是一个理想的 Web 框架. 一.Tornado的两种模式使用 1.同步阻塞模式 由于doing中sleep10秒,此时其他连接将被阻塞,必须等这次请求完成后其他请求才能连接成功. 1 import tornado.io

利用tornado使请求实现异步非阻塞

基本IO模型 网上搜了很多关于同步异步,阻塞非阻塞的说法,理解还是不能很透彻,有必要买书看下. 参考:使用异步 I/O 大大提高应用程序的性能 怎样理解阻塞非阻塞与同步异步的区别? 同步和异步:主要关注消息通信机制(重点在B?). 同步:A调用B,B处理直到获得结果,才返回给A. 异步:A调用B,B直接返回.无需等待结果,B通过状态,通知等来通知A或回调函数来处理. 阻塞和非阻塞:主要关注程序等待(重点在A?). 阻塞:A调用B,A被挂起直到B返回结果给A,A继续执行. 非阻塞:A调用B,A不会

使用tornado让你的请求异步非阻塞

http://www.dongwm.com/archives/shi-yong-tornadorang-ni-de-qing-qiu-yi-bu-fei-zu-sai/?utm_source=tuicool&utm_medium=referral 前言 也许有同学很迷惑:tornado不是标榜异步非阻塞解决10K问题的嘛?但是我却发现不是torando不好,而是你用错了.比如最近发现一个事情:某网站打开页面很慢,服务器cpu/内存都正常.网络状态也良好. 后来发现,打开页面会有很多请求后端数据库

CompletionService异步非阻塞获取并行任务执行结果

第1部分 问题引入 <Java并发编程实践>一书6.3.5节CompletionService:Executor和BlockingQueue,有这样一段话: "如果向Executor提交了一组计算任务,并且希望在计算完成后获得结果,那么可以保留与每个任务关联的Future,然后反复使用get方法,同时将参数timeout指定为0,从而通过轮询来判断任务是否完成.这种方法虽然可行,但却有些繁琐.幸运的是,还有一种更好的方法:完成服务CompletionService." 这是

200行自定义异步非阻塞Web框架

Python的Web框架中Tornado以异步非阻塞而闻名.本篇将使用200行代码完成一个微型异步非阻塞Web框架:Snow. 一.源码 本文基于非阻塞的Socket以及IO多路复用从而实现异步非阻塞的Web框架,其中便是众多异步非阻塞Web框架内部原理. #!/usr/bin/env python # -*- coding:utf-8 -*- import re import socket import select import time class HttpResponse(object)