摘要：定義一個(gè)閉包相關(guān)官方文檔是一種用于并發(fā)編程的模式，首次引入是在的模塊。對(duì)象是一個(gè)對(duì)于異步返回結(jié)果的占位符。一個(gè)對(duì)象包含了一次異步操作的結(jié)果。在同步編程中，被用于等待從一個(gè)線程池或進(jìn)程池里返回的結(jié)果在中，通常被用在或者在一個(gè)函數(shù)中它們。

from tornado.concurrent import Future

def async_call_method(fun, *args, **kwargs):
    future = Future()
    // 定義一個(gè)閉包 finish
    def finish():
        try:
            result = fun(*args, **kwargs)
            if future._callbacks:
                IOLoop.current().add_callback(future.set_result, result)
            else:
                future.set_result(result)
        except:
            if future._callbacks:
                IOLoop.current().add_callback(future.set_exc_info, sys.exc_info())
            else:
                future.set_exc_info(sys.exc_info())
    child_gr = greenlet.greenlet(finish)
    child_gr.switch()
    return future

Future 是一種用于并發(fā)編程的模式，首次引入是在 python 3.2 的 concurrent.futures 模塊。

Future 對(duì)象是一個(gè)對(duì)于異步返回結(jié)果的占位符。

一個(gè) Future 對(duì)象包含了一次異步操作的結(jié)果。在同步編程中，F(xiàn)utures 被用于等待從一個(gè)線程池或進(jìn)程池里返回的結(jié)果；在 tornado 中，future 通常被用在 IOLoop.add_future 或者在一個(gè) gen.coroutine 函數(shù)中 yielding 它們。

tornado.concurrent.Future 和 concurrent.futures.Future 相似，但是其不是線程安全的(因此，在單線程事件循環(huán)應(yīng)用在速度更快)

經(jīng)過一番搜索，查詢到 async_call_method() 這個(gè)函數(shù)來自于 github.com/snower/TorMySQL.

經(jīng)過對(duì)該項(xiàng)目代碼的仔細(xì)閱讀，我發(fā)現(xiàn)了它是如何實(shí)現(xiàn)了 mysql 的異步操作。

...
def connect(self):
    # 入口函數(shù)
    # 設(shè)置 future 占位符
    future = Future()

    # 定義回調(diào)函數(shù)
    def on_connected(connection_future):
        if connection_future._exc_info is None:
            future.set_result(self)
        else:
            future.set_exc_info(connection_future.exc_info())
    self._connection = Connection(defer_connect = True, *self._args, **self._kwargs)
    self._connection.set_close_callback(self.connection_close_callback)

    # 用 greenlet 包裝 self._connection.connect 并返回 future
    # 要使 async_call_method 包裝后的函數(shù)有非阻塞的特性，必須達(dá)成以下要求
    # 1. 函數(shù)可以訪問 父greenlet
    # 2. 函數(shù)中所有 IO 操作均支持非阻塞(比如: 非阻塞由 socket 的 non-blocking 特性支持)
    # 3. 函數(shù)中執(zhí)行 IO 操作后立即將運(yùn)行權(quán)交還給主函數(shù)(父greenlet, 如：ioloop 時(shí)間循環(huán))(greenlet.switch)
    # 4. 函數(shù)中所有 IO 操作均返回 Future
    # 5. Future.callback 運(yùn)行后立即將運(yùn)行權(quán)(greenlet.switch)返回給當(dāng)前函數(shù)(greenlet.current)，完成當(dāng)前函數(shù)的剩余部分
    connection_future = async_call_method(self._connection.connect)

    # 當(dāng) connection_future 狀態(tài)為 finished, 調(diào)用 on_connected()
    # finished => 調(diào)用 connection_future.set_result()
    IOLoop.current().add_future(connection_future, on_connected)
    return future
...

...
        # IOStream 基于 tornado.iostream.IOStream
        sock = IOStream(sock)
        sock.set_close_callback(self.stream_close_callback)

        # getcurrent() 返回包裝了當(dāng)前函數(shù)的 greenlet
        child_gr = greenlet.getcurrent()
        # main 是指 父greenlet(主函數(shù), 時(shí)間循環(huán)?)
        main = child_gr.parent
        assert main is not None, "Execut must be running in child greenlet"
...
    def connected(future):
        if self._loop_connect_timeout:
            self._loop.remove_timeout(self._loop_connect_timeout)
            self._loop_connect_timeout = None
    
        if future._exc_info is not None:
            child_gr.throw(future.exception())
        else:
            self._sock = sock
            # 將運(yùn)行權(quán)交還給當(dāng)前 greenlet
            child_gr.switch()
    
    # IOStream.connect 是 no-blocking 的 socket 操作
    future = sock.connect(address)
    # 給 sock.connect 操作添加回調(diào)函數(shù)
    self._loop.add_future(future, connected)
    # 然后把運(yùn)行權(quán)交還給 父greenlet
    # 直到連接成功，connected() 中會(huì)將運(yùn)行權(quán)交還給 當(dāng)前greenlet
    main.switch()
...

要使 async_call_method 包裝后的函數(shù)有非阻塞的特性，必須達(dá)成以下要求

函數(shù)可以訪問 父greenlet

函數(shù)中所有 IO 操作均支持非阻塞(比如: 非阻塞由 socket 的 non-blocking 特性支持)

函數(shù)中執(zhí)行 IO 操作后立即將運(yùn)行權(quán)交還給主函數(shù)(父greenlet, 如：ioloop 時(shí)間循環(huán))(greenlet.switch)

函數(shù)中所有 IO 操作均返回 Future

Future.callback 運(yùn)行后立即將運(yùn)行權(quán)(greenlet.switch)返回給當(dāng)前函數(shù)(greenlet.current)，完成當(dāng)前函數(shù)的剩余部分

async_call_method 包裝后的函數(shù)要實(shí)現(xiàn)非阻塞，最終還是依賴于 socket 的非阻塞
=> socket.setblocking(False)。

github.com/snower/TorMySQL 中于 mysql 的交互全部通過 IOStream() 的以下方法實(shí)現(xiàn):

* def _handle_events(self, fd, events):  # ioloop 在事件發(fā)生時(shí)調(diào)用 _handle_events
* def _handle_connect(self): 
* def _handle_read(self):  # 當(dāng)事件為讀取事件時(shí)，讀取數(shù)據(jù)到 buffer, 然后 future.set_result(data)
* def _handle_write(self):  # 當(dāng)事件為寫事件時(shí)，讀取數(shù)據(jù)到 buffer, 然后 future.set_result(data)
* def read(self, num_bytes):
* def write(self, data):

通過對(duì)上述方法進(jìn)行 設(shè)置 future 占位符，并基于 non-blocking socket 實(shí)現(xiàn)上述方法的非阻塞。