什么意思呢？就是說，tornado：老子只給你提供異步的入口，你要是真想異步操作，要不你就使用我提供的一些異步客戶端來搞，不然你就自己實現一個異步的操作。

以操作MongoDB為例，如果你的函數中含有調用mongo的調用（使用pymongo庫），那么這時候你加asynchronous這個裝飾器就沒有任何效果了，因為你的mongo調用本身是同步的，如果想做成異步非阻塞的效果，需要使用mongo出品的另一個python driver -- motor，這個driver支持異步操作mongo，這時候你再加asynchronous裝飾器并操作mongo就可以實現異步非阻塞的效果了。

所以，如果要使用tornado的異步調用，第一，使用tornado內置的異步客戶端如httpclient.AsyncHTTPClient等；第二，可參考內置異步客戶端，借助tornado.ioloop.IOLoop封裝一個自己的異步客戶端，但開發(fā)成本并不小。

然而，天無絕人之路，還是有辦法可以用較低的成本實現tornado的異步非阻塞的，那就是借助celery項目。前面說了，它是一個分布式的實時處理消息隊列調度系統(tǒng)，tornado接到請求后，可以把所有的復雜業(yè)務邏輯處理、數據庫操作以及IO等各種耗時的同步任務交給celery，由這個任務隊列異步處理完后，再返回給tornado。這樣只要保證tornado和celery的交互是異步的，那么整個服務是完全異步的。至于如何保證tornado和celery之間的交互是異步的，可以借助tornado-celery這個適配器來實現。

celery配合rabbitmq的工作流程如下：

這里我們來使用這幾個組件重寫前面的同步阻塞的例子：

#!/bin/env python

import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
import tornado.httpclient

import time
import tcelery, tasks
from tornado.options import define, options
tcelery.setup_nonblocking_producer()
define("port", default=8000, help="run on the given port", type=int)

class AsyncMysqlHandler(tornado.web.RequestHandler):
    @tornado.web.asynchronous
    @tornado.gen.coroutine
    def get(self, flag):
        res = yield tornado.gen.Task(tasks.query_mysql.apply_async, args=[flag])
        self.write(res.result)
        self.finish()

class NowHandler(tornado.web.RequestHandler):
    def get(self):
        self.write("i want you, right now!")

if __name__ == "__main__":
    tornado.options.parse_command_line()
    app = tornado.web.Application(handlers=[
            (r"/mysql_query/(d+)", AsyncMysqlHandler), 
            (r"/i_want_you_now", NowHandler)])
    http_server = tornado.httpserver.HTTPServer(app)
    http_server.listen(options.port)
    tornado.ioloop.IOLoop.instance().start()

這里有個新的tornado.gen.coroutine裝飾器, coroutine是3.0之后新增的裝飾器.以前的辦法是用回調函數的方式進行異步調用，如果使用回調函數的方式，則代碼如下：

#!/bin/env python

import tornado.httpserver
import tornado.ioloop
import tornado.options
import tornado.web
import tornado.httpclient
import time
import tcelery, tasks
from tornado.options import define, options
tcelery.setup_nonblocking_producer()
define("port", default=8000, help="run on the given port", type=int)

class AsyncMysqlHandler(tornado.web.RequestHandler):
    @tornado.web.asynchronous
    def get(self, flag):
        tasks.query_mysql.apply_async(args=[flag], callback=self.on_result)

    def on_result(self, response):
        self.write(response.result)
        self.finish()

class NowHandler(tornado.web.RequestHandler):
    def get(self):
        self.write("i want you, right now!")

if __name__ == "__main__":
    tornado.options.parse_command_line()
    app = tornado.web.Application(handlers=[
            (r"/mysql_query/(d+)", AsyncMysqlHandler), 
            (r"/i_want_you_now", NowHandler)])
    http_server = tornado.httpserver.HTTPServer(app)
    http_server.listen(options.port)
    tornado.ioloop.IOLoop.instance().start()

使用callback的話始終覺得會是的代碼結構變得比較混亂，試想如果有大量異步回調，每一個都寫一個回調函數的話，勢必導致項目代碼結構變得不那么清晰和優(yōu)雅，畢竟回調這種反人類的寫法還是很多人不喜歡的，但也看個人喜好，不喜歡callback風格的可以使用yield來進行異步調用。

tasks.py集中放置開發(fā)者需要異步執(zhí)行的函數。

import time
import torndb
from celery import Celery

db = torndb.Connection("127.0.0.1:3306", "user_db", "username", "passwd")
app = Celery("tasks", broker="amqp://guest:guest@localhost:5672")
app.conf.CELERY_RESULT_BACKEND = "amqp://guest:guest@localhost:5672"

@app.task(name="task.query_users")
def query_mysql(flag):
    return db.query("select * from table where flag=%s", flag)

if __name__ == "__main__":
    app.start()

然后啟動celery worker監(jiān)聽任務隊列（消費者會從任務隊列中取走一個個的task并執(zhí)行）：

celery -A tasks worker --loglevel=info

自此，依靠這種架構，可以實現tornado處理請求的完全異步調用。

使用上述方案的異步非阻塞可能會依賴于celery的任務隊列長度，若隊列中的任務過多，則可能導致長時間等待，降低效率。
解決方案：

啟動多個celery worker監(jiān)聽任務隊列，使用多進程并發(fā)消費任務隊列，celery命令可以通過-concurrency參數來指定用來執(zhí)行任務而prefork的worker進程，如果所有的worker都在執(zhí)行任務，那么新添加的任務必須要等待有一個正在執(zhí)行的任務完成后才能被執(zhí)行，默認的concurrency數量是機器上CPU的數量。另外，celery是支持好幾個并發(fā)模式的，有prefork，threading，協(xié)程（gevent，eventlet），prefork在celery的介紹是，默認是用了multiprocess來實現的；可以通過-p參數指定其他的并發(fā)模型，如gevent（需自己配置好gevent環(huán)境）。

建立多個任務queue，把大量的任務分發(fā)到不同的queue中，減輕單個queue時可能出現的任務數量過載。

前面說了celery是一個分布式系統(tǒng)，也就是說，基于celery的項目可無痛實現分布式擴展，前面寫的tornado和celery配合的demo，也可以實現獨立部署，即tornado server和celery server其實可以分開部署，即分布在不同的服務器上，celery server部署自己的tasks.py任務，并啟動celery worker監(jiān)聽，然后在tornado server上添加以下代碼：

from celery import Celery
app = Celery(broker = "amqp://",)

并使用Celery的send_task函數調用任務：

app.send_task("function_name", args=[param1, param2, param3...])

即可實現tornado和celery的完全解耦。

另外，了解到tornado.concurrent.futures（py3自帶這個庫，py2需多帶帶安裝）這個module可以實現自定義函數的異步化，目前還沒有深入了解這個東西，有時間去研究一下這個東西，有心得再分享一下這個module相關的知識。