tornado異步請求非阻塞

2019-11-14 17:52:00

字體：大中小

來源：轉載

供稿：網友

前言

也許有同學很迷惑:tornado不是標榜異步非阻塞解決10K問題的嘛?但是我卻發現不是torando不好，而是你用錯了.比如最近發現一個事情:某網站打開頁面很慢,服務器cpu/內存都正常.網絡狀態也良好. 后來發現，打開頁面會有很多請求后端數據庫的訪問，有一個mongodb的數據庫業務api的rest服務.但是它的tornado卻用錯了,一步步的來研究問題:

說明

以下的例子都有2個url,一個是耗時的請求，一個是可以或者說需要立刻返回的請求,我想就算一個對技術不熟，從道理上來說的用戶，他希望的是他訪問的請求不會影響也不會被其他人的請求影響

#!/bin/env pythonimport tornado.httpserverimport tornado.ioloopimport tornado.optionsimport tornado.webimport tornado.httpclientimport timefrom tornado.options import define, optionsdefine("port", default=8000, help="run on the given port", type=int)class SleepHandler(tornado.web.RequestHandler):    def get(self):        time.sleep(5)        self.write("when i sleep 5s")class JustNowHandler(tornado.web.RequestHandler):    def get(self):        self.write("i hope just now see you")if __name__ == "__main__":    tornado.options.parse_command_line()    app = tornado.web.application(handlers=[            (r"/sleep", SleepHandler), (r"/justnow", JustNowHandler)])    http_server = tornado.httpserver.HTTPServer(app)    http_server.listen(options.port)    tornado.ioloop.IOLoop.instance().start()

假如你使用頁面請求或者使用哪個httpie,curl等工具先訪問http://localhost:8000/sleep,再訪問http://localhost:8000/justnow.你會發現本來可以立刻返回的/jsutnow的請求會一直阻塞到/sleep請求完才返回.

這是為啥?為啥我的請求被/sleep請求阻塞了？如果平時我們的web請求足夠快我們可能不會意識到這個問題，但是事實上經常會有一些耗時的進程，意味著應用程序被有效的鎖定直至處理結束.

這是時候你有沒有想起@tornado.web.asynchronous這個裝飾器？但是使用這個裝飾器有個前提就是你要耗時的執行需要執行異步,比如上面的time.sleep,你只是加裝飾器是沒有作用的，而且需要注意的是 Tornado默認在函數處理返回時關閉客戶端的連接,但是當你使用@tornado.web.asynchonous裝飾器時，Tornado永遠不會自己關閉連接，需要顯式的self.finish()關閉

我們大部分的函數都是阻塞的, 比如上面的time.sleep其實tornado有個異步的實現:

#!/bin/env pythonimport tornado.httpserverimport tornado.ioloopimport tornado.optionsimport tornado.webimport tornado.genimport tornado.httpclientimport tornado.concurrentimport tornado.ioloopimport timefrom tornado.options import define, optionsdefine("port", default=8000, help="run on the given port", type=int)class SleepHandler(tornado.web.RequestHandler):    @tornado.web.asynchronous    @tornado.gen.coroutine    def get(self):        yield tornado.gen.Task(tornado.ioloop.IOLoop.instance().add_timeout, time.time() + 5)        self.write("when i sleep 5s")class JustNowHandler(tornado.web.RequestHandler):    def get(self):        self.write("i hope just now see you")if __name__ == "__main__":    tornado.options.parse_command_line()    app = tornado.web.Application(handlers=[            (r"/sleep", SleepHandler), (r"/justnow", JustNowHandler)])    http_server = tornado.httpserver.HTTPServer(app)    http_server.listen(options.port)    tornado.ioloop.IOLoop.instance().start()

這里有個新的tornado.gen.coroutine裝飾器, coroutine是3.0之后新增的裝飾器.以前的辦法是用回調，還是看我這個例子:

class SleepHandler(tornado.web.RequestHandler):    @tornado.web.asynchronous    def get(self):        tornado.ioloop.IOLoop.instance().add_timeout(time.time() + 5, callback=self.on_response)    def on_response(self):        self.write("when i sleep 5s")        self.finish()

使用了callback, 但是新的裝飾器讓我們通過yield實現同樣的效果:你在打開/sleep之后再點擊/justnow， justnow的請求都是立刻返回不受影響.但是用了asynchronous的裝飾器你的耗時的函數也需要執行異步

剛才說的都是沒有意義的例子，下面寫個有點用的:讀取mongodb數據庫數據，然后再前端按行write出來

#!/bin/env pythonimport tornado.httpserverimport tornado.ioloopimport tornado.optionsimport tornado.webimport tornado.genimport tornado.httpclientimport tornado.concurrentimport tornado.ioloopimport time# 一個mongodb出品的支持異步的數據庫的python驅動import motorfrom tornado.options import define, optionsdefine("port", default=8000, help="run on the given port", type=int)# db其實就是test數據庫的游標db = motor.MotorClient().open_sync().testclass SleepHandler(BaseHandler):    @tornado.web.asynchronous    @tornado.gen.coroutine    def get(self):        # 這一行執行還是阻塞需要時間的，我的tt集合有一些數據并且沒有索引        cursor = db.tt.find().sort([('a', -1)])        # 這部分會異步非阻塞的執行二不影響其他頁面請求        while (yield cursor.fetch_next):            message = cursor.next_object()            self.write('<li>%s</li>' % message['a'])        self.write('</ul>')        self.finish()    def _on_response(self, message, error):        if error:            raise tornado.web.HTTPError(500, error)        elif message:            for i in message:                self.write('<li>%s</li>' % i['a'])        else:            self.write('</ul>')            self.finish()class JustNowHandler(BaseHandler):    def get(self):        self.write("i hope just now see you")if __name__ == "__main__":    tornado.options.parse_command_line()    app = tornado.web.Application(handlers=[            (r"/sleep", SleepHandler), (r"/justnow", JustNowHandler)])    http_server = tornado.httpserver.HTTPServer(app)    http_server.listen(options.port)    tornado.ioloop.IOLoop.instance().start()

一個同事提示為什么這個耗時的東西不能異步的丟給某工具去執行而不阻塞我的請求呢?好吧，我也想到了:celery，正好github有這個東西:tornado-celery

執行下面的程序首先你要安裝rabbitmq和celery:

#!/bin/env pythonimport tornado.httpserverimport tornado.ioloopimport tornado.optionsimport tornado.webimport tornado.genimport tornado.httpclientimport tcelery, tasksimport timefrom tornado.options import define, optionsdefine("port", default=8000, help="run on the given port", type=int)tcelery.setup_nonblocking_PRoducer()class SleepHandler(tornado.web.RequestHandler):    @tornado.web.asynchronous    @tornado.gen.coroutine    def get(self):        # tornado.gen.Task的參數是:要執行的函數, 參數        yield tornado.gen.Task(tasks.sleep.apply_async, args=[5])        self.write("when i sleep 5s")        self.finish()class JustNowHandler(tornado.web.RequestHandler):    def get(self):        self.write("i hope just now see you")if __name__ == "__main__":    tornado.options.parse_command_line()    app = tornado.web.Application(handlers=[            (r"/sleep", SleepHandler), (r"/justnow", JustNowHandler)])    http_server = tornado.httpserver.HTTPServer(app)    http_server.listen(options.port)    tornado.ioloop.IOLoop.instance().start()

task是celery的任務定義的文件，包含我們說的time.sleep的函數

import timefrom celery import Celerycelery = Celery("tasks", broker="amqp://guest:guest@localhost:5672")celery.conf.CELERY_RESULT_BACKEND = "amqp"@celery.taskdef sleep(seconds):    time.sleep(float(seconds))    return secondsif __name__ == "__main__":    celery.start()

然后啟動celelry worker(要不然你的任務怎么執行呢?肯定需要一個消費者取走):

celery -A tasks worker --loglevel=info

但是這里的問題也可能很嚴重:我們的異步非阻塞依賴于celery,還是這個隊列的長度,假如任務很多那么就需要等待,效率很低.有沒有一種辦法把我的同步阻塞函數變為異步(或者說被tornado的裝飾器理解和識別)呢?

#!/bin/env pythonimport tornado.httpserverimport tornado.ioloopimport tornado.optionsimport tornado.webimport tornado.httpclientimport tornado.genfrom tornado.concurrent import run_on_executor# 這個并發庫在python3自帶在python2需要安裝sudo pip install futuresfrom concurrent.futures import ThreadPoolExecutorimport timefrom tornado.options import define, optionsdefine("port", default=8000, help="run on the given port", type=int)class SleepHandler(tornado.web.RequestHandler):    executor = ThreadPoolExecutor(2)　　#executor 是局部變量  不是全局的
    @tornado.web.asynchronous    @tornado.gen.coroutine    def get(self):        # 假如你執行的異步會返回值被繼續調用可以這樣(只是為了演示),否則直接yield就行        res = yield self.sleep()        self.write("when i sleep %s s" % res)        self.finish()    @run_on_executor    def sleep(self):        time.sleep(5)        return 5class JustNowHandler(tornado.web.RequestHandler):    def get(self):        self.write("i hope just now see you")if __name__ == "__main__":    tornado.options.parse_command_line()    app = tornado.web.Application(handlers=[            (r"/sleep", SleepHandler), (r"/justnow", JustNowHandler)])    http_server = tornado.httpserver.HTTPServer(app)    http_server.listen(options.port)    tornado.ioloop.IOLoop.instance().start()
本文轉載自http://www.dongwm.com/archives/shi-yong-tornadorang-ni-de-qing-qiu-yi-bu-fei-zu-sai/