tornado异步请求非阻塞
前言
也許有同學(xué)很迷惑:tornado不是標(biāo)榜異步非阻塞解決10K問題的嘛?但是我卻發(fā)現(xiàn)不是torando不好，而是你用錯(cuò)了.比如最近發(fā)現(xiàn)一個(gè)事情:某網(wǎng)站打開頁面很慢,服務(wù)器cpu/內(nèi)存都正常.網(wǎng)絡(luò)狀態(tài)也良好. 后來發(fā)現(xiàn)，打開頁面會(huì)有很多請(qǐng)求后端數(shù)據(jù)庫的訪問，有一個(gè)mongodb的數(shù)據(jù)庫業(yè)務(wù)api的rest服務(wù).但是它的tornado卻用錯(cuò)了,一步步的來研究問題:
?
說明
以下的例子都有2個(gè)url,一個(gè)是耗時(shí)的請(qǐng)求，一個(gè)是可以或者說需要立刻返回的請(qǐng)求,我想就算一個(gè)對(duì)技術(shù)不熟，從道理上來說的用戶， 他希望的是他訪問的請(qǐng)求不會(huì)影響也不會(huì)被其他人的請(qǐng)求影響
?
#!/bin/env python
import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
import tornado.httpclient
import time
from tornado.options import define, options
define("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()
假如你使用頁面請(qǐng)求或者使用哪個(gè)httpie,curl等工具先訪問http://localhost:8000/sleep,再訪問http://localhost:8000/justnow.你會(huì)發(fā)現(xiàn)本來可以立刻返回的/jsutnow的請(qǐng)求會(huì)一直阻塞到/sleep請(qǐng)求完才返回.
?
這是為啥?為啥我的請(qǐng)求被/sleep請(qǐng)求阻塞了？如果平時(shí)我們的web請(qǐng)求足夠快我們可能不會(huì)意識(shí)到這個(gè)問題，但是事實(shí)上經(jīng)常會(huì)有一些耗時(shí)的進(jìn)程，意味著應(yīng)用程序被有效的鎖定直至處理結(jié)束.
?
這是時(shí)候你有沒有想起@tornado.web.asynchronous這個(gè)裝飾器？但是使用這個(gè)裝飾器有個(gè)前提就是你要耗時(shí)的執(zhí)行需要執(zhí)行異步,比如上面的time.sleep,你只是加裝飾器是沒有作用的，而且需要注意的是 Tornado默認(rèn)在函數(shù)處理返回時(shí)關(guān)閉客戶端的連接,但是當(dāng)你使用@tornado.web.asynchonous裝飾器時(shí)，Tornado永遠(yuǎn)不會(huì)自己關(guān)閉連接，需要顯式的self.finish()關(guān)閉
?
我們大部分的函數(shù)都是阻塞的, 比如上面的time.sleep其實(shí)tornado有個(gè)異步的實(shí)現(xiàn):
?
#!/bin/env python
import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
import tornado.gen
import tornado.httpclient
import tornado.concurrent
import tornado.ioloop
import time
from tornado.options import define, options
define("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()
這里有個(gè)新的tornado.gen.coroutine裝飾器, coroutine是3.0之后新增的裝飾器.以前的辦法是用回調(diào)，還是看我這個(gè)例子:
?
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實(shí)現(xiàn)同樣的效果:你在打開/sleep之后再點(diǎn)擊/justnow， justnow的請(qǐng)求都是立刻返回不受影響.但是用了asynchronous的裝飾器你的耗時(shí)的函數(shù)也需要執(zhí)行異步
?
剛才說的都是沒有意義的例子，下面寫個(gè)有點(diǎn)用的:讀取mongodb數(shù)據(jù)庫數(shù)據(jù)，然后再前端按行write出來
?
#!/bin/env python
import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
import tornado.gen
import tornado.httpclient
import tornado.concurrent
import tornado.ioloop
import time
# 一個(gè)mongodb出品的支持異步的數(shù)據(jù)庫的python驅(qū)動(dòng)
import motor
from tornado.options import define, options
define("port", default=8000, help="run on the given port", type=int)
# db其實(shí)就是test數(shù)據(jù)庫的游標(biāo)
db = motor.MotorClient().open_sync().test
class SleepHandler(BaseHandler):
? ? @tornado.web.asynchronous
? ? @tornado.gen.coroutine
? ? def get(self):
? ? ? ? # 這一行執(zhí)行還是阻塞需要時(shí)間的，我的tt集合有一些數(shù)據(jù)并且沒有索引
? ? ? ? cursor = db.tt.find().sort([('a', -1)])
? ? ? ? # 這部分會(huì)異步非阻塞的執(zhí)行二不影響其他頁面請(qǐng)求
? ? ? ? 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()
一個(gè)同事提示為什么這個(gè)耗時(shí)的東西不能異步的丟給某工具去執(zhí)行而不阻塞我的請(qǐng)求呢?好吧，我也想到了:celery，正好github有這個(gè)東西:tornado-celery
?
執(zhí)行下面的程序首先你要安裝rabbitmq和celery:
?
#!/bin/env python
import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
import tornado.gen
import tornado.httpclient
import tcelery, tasks
import time
from tornado.options import define, options
define("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的參數(shù)是:要執(zhí)行的函數(shù), 參數(shù)
? ? ? ? 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的任務(wù)定義的文件，包含我們說的time.sleep的函數(shù)
?
import time
from celery import Celery
celery = Celery("tasks", broker="amqp://guest:guest@localhost:5672")
celery.conf.CELERY_RESULT_BACKEND = "amqp"
@celery.task
def sleep(seconds):
? ? time.sleep(float(seconds))
? ? return seconds
if __name__ == "__main__":
? ? celery.start()
然后啟動(dòng)celelry worker(要不然你的任務(wù)怎么執(zhí)行呢?肯定需要一個(gè)消費(fèi)者取走):
?
celery -A tasks worker --loglevel=info
但是這里的問題也可能很嚴(yán)重:我們的異步非阻塞依賴于celery,還是這個(gè)隊(duì)列的長(zhǎng)度,假如任務(wù)很多那么就需要等待,效率很低.有沒有一種辦法把我的同步阻塞函數(shù)變?yōu)楫惒?或者說被tornado的裝飾器理解和識(shí)別)呢?
?
#!/bin/env python
import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
import tornado.httpclient
import tornado.gen
from tornado.concurrent import run_on_executor
# 這個(gè)并發(fā)庫在python3自帶在python2需要安裝sudo pip install futures
from concurrent.futures import ThreadPoolExecutor
import time
from tornado.options import define, options
define("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):
? ? ? ? # 假如你執(zhí)行的異步會(huì)返回值被繼續(xù)調(diào)用可以這樣(只是為了演示),否則直接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 5
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()
總結(jié)
以上是生活随笔為你收集整理的tornado异步请求非阻塞的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問題。
- 上一篇： 使用python连接数据库
- 下一篇： HDU 1848 Fibonacci a