摘要：主程序通過喚起子程序并傳入數據，子程序處理完后，用將自己掛起，并返回主程序，如此交替進行。通過輪詢或是等事件框架，捕獲返回的事件。從消息隊列中取出記錄，恢復協程函數。然而事實上只有直接操縱的協程函數才有可能接觸到這個對象。

首發于 我的博客 轉載請注明出處

本文默認讀者對 Python 生成器 有一定的了解，不了解者請移步至生成器 - 廖雪峰的官方網站。

本文基于 Python 3.5.1，文中所有的例子都可在 Github 上獲得。

學過 Python 的都知道，Python 里有一個很厲害的概念叫做 生成器（Generators）。一個生成器就像是一個微小的線程，可以隨處暫停，也可以隨時恢復執行，還可以和代碼塊外部進行數據交換。恰當使用生成器，可以極大地簡化代碼邏輯。

也許，你可以熟練地使用生成器完成一些看似不可能的任務，如“無窮斐波那契數列”，并引以為豪，認為所謂的生成器也不過如此——那我可要告訴你：這些都太小兒科了，下面我所要介紹的絕對會讓你大開眼界。

生成器 可以實現 協程，你相信嗎？

在異步編程盛行的今天，也許你已經對 協程（coroutines） 早有耳聞，但卻不一定了解它。我們先來看看 Wikipedia 的定義：

Coroutines are computer program components that generalize subroutines for nonpreemptive multitasking, by allowing multiple entry points for suspending and resuming execution at certain locations.

也就是說：協程是一種 允許在特定位置暫停或恢復的子程序——這一點和 生成器 相似。但和 生成器 不同的是，協程 可以控制子程序暫停之后代碼的走向，而 生成器 僅能被動地將控制權交還給調用者。

協程 是一種很實用的技術。和 多進程 與 多線程 相比，協程 可以只利用一個線程更加輕便地實現 多任務，將任務切換的開銷降至最低。和 回調 等其他異步技術相比，協程 維持了正常的代碼流程，在保證代碼可讀性的同時最大化地利用了 阻塞 IO 的空閑時間。它的高效與簡潔贏得了開發者們的擁戴。

早先 Python 是沒有原生協程支持的，因此在 協程 這個領域出現了百家爭鳴的現象。主流的實現由以下兩種：

用 C 實現協程調度。這一派以 gevent 為代表，在底層實現了協程調度，并將大部分的 阻塞 IO 重寫為異步。

用 生成器模擬。這一派以 Tornado 為代表。Tornado 是一個老牌的異步 Web 框架，涵蓋了五花八門的異步編程方式，其中包括 協程。本文部分代碼借鑒于 Tornado。

直至 Python 3.4，Python 第一次將異步編程納入標準庫中（參見 PEP 3156），其中包括了用生成器模擬的 協程。而在 Python 3.5 中，Guido 總算在語法層面上實現了 協程（參見 PEP 0492）。比起 yield 關鍵字，新關鍵字 async 和 await 具有更好的可讀性。在不久的將來，新的實現將會慢慢統一混亂已久的協程領域。

盡管 生成器協程 已成為了過去時，但它曾經的輝煌卻不可磨滅。下面，讓我們一起來探索其中的魔法。

假設有兩個子程序 main 和 printer。printer 是一個死循環，等待輸入、加工并輸出結果。main 作為主程序，不時地向 printer 發送數據。

這應該怎么實現呢？

傳統方式中，這幾乎不可能在一個線程中實現，因為死循環會阻塞。而協程卻能很好地解決這個問題：

def printer():

    counter = 0
    while True:
        string = (yield)
        print("[{0}] {1}".format(counter, string))
        counter += 1

if __name__ == "__main__":
    p = printer()
    next(p)
    p.send("Hi")
    p.send("My name is hsfzxjy.")
    p.send("Bye!")

輸出：

[0] Hi
[1] My name is hsfzxjy.
[2] Bye!

這其實就是最簡單的協程。程序由兩個分支組成。主程序通過 send 喚起子程序并傳入數據，子程序處理完后，用 yield 將自己掛起，并返回主程序，如此交替進行。

有時，你的手頭上會有多個任務，每個任務耗時很長，而你又不想同步處理，而是希望能像多線程一樣交替執行。這時，你就需要一個調度器來協調流程了。

作為例子，我們假設有這么一個任務：

def task(name, times):

    for i in range(times):
        print(name, i)

如果你直接執行 task，那它會在遍歷 times 次之后才會返回。為了實現我們的目的，我們需要將 task 人為地切割成若干塊，以便并行處理：

def task(name, times):

    for i in range(times):
        yield
        print(name, i)

這里的 yield 沒有邏輯意義，僅是作為暫停的標志點。程序流可以在此暫停，也可以在此恢復。而通過實現一個調度器，我們可以完成多個任務的并行處理：

from collections import deque

class Runner(object):

    def __init__(self, tasks):
        self.tasks = deque(tasks)

    def next(self):
        return self.tasks.pop()

    def run(self):
        while len(self.tasks):
            task = self.next()
            try:
                next(task)
            except StopIteration:
                pass
            else:
                self.tasks.appendleft(task)

這里我們用一個隊列（deque）儲存任務列表。其中的 run 是一個重要的方法： 它通過輪轉隊列依次喚起任務，并將已經完成的任務清出隊列，簡潔地模擬了任務調度的過程。

而現在，我們只需調用：

Runner([
    task("hsfzxjy", 5),
    task("Jack", 4),
    task("Bob", 6)
]).run()

就可以得到預想中的效果了：

Bob 0
Jack 0
hsfzxjy 0
Bob 1
Jack 1
hsfzxjy 1
Bob 2
Jack 2
hsfzxjy 2
Bob 3
Jack 3
hsfzxjy 3
Bob 4
hsfzxjy 4
Bob 5

簡直完美！答案和丑陋的多線程別無二樣，代碼卻簡單了不止一個數量級。

你絕對有過這樣的煩惱：程序常常被時滯嚴重的 IO 操作（數據庫查詢、大文件讀取、越過長城拿數據）阻塞，在等待 IO 返回期間，線程就像死了一樣，空耗著時間。為此，你不得不用多線程甚至是多進程來解決問題。

而事實上，在等待 IO 的時候，你完全可以做一些與數據無關的操作，最大化地利用時間。Node.js 在這點做得不錯——它將一切異步化，壓榨性能。只可惜它的異步是基于事件回調機制的，稍有不慎，你就有可能陷入 Callback Hell 的深淵。

而協程并不使用回調，相比之下可讀性會好很多。其思路大致如下：

維護一個消息隊列，用于儲存 IO 記錄。

協程函數 IO 時，自身掛起，同時向消息隊列插入一個記錄。

通過輪詢或是 epoll 等事件框架，捕獲 IO 返回的事件。

從消息隊列中取出記錄，恢復協程函數。

現在假設有這么一個耗時任務：

def task(name):
    print(name, 1)
    sleep(1)
    print(name, 2)
    sleep(2)
    print(name, 3)

正常情況下，這個任務執行完需要 3 秒，倘若多個同步任務同步執行，執行時間會成倍增長。而如果利用協程，我們就可以在接近 3 秒的時間內完成多個任務。

首先我們要實現消息隊列：

events_list = []


class Event(object):

    def __init__(self, *args, **kwargs):
        self.callback = lambda: None
        events_list.append(self)

    def set_callback(self, callback):
        self.callback = callback

    def is_ready(self):
        result = self._is_ready()

        if result:
            self.callback()

        return result

Event 是消息的基類，其在初始化時會將自己放入消息隊列 events_list 中。Event 和 調度器 使用回調進行交互。

接著我們要 hack 掉 sleep 函數，這是因為原生的 time.sleep() 會阻塞線程。通過自定義 sleep 我們可以模擬異步延時操作：

# sleep.py

from event import Event
from time import time


class SleepEvent(Event):

    def __init__(self, timeout):
        super(SleepEvent, self).__init__(timeout)
        self.timeout = timeout
        self.start_time = time()

    def _is_ready(self):
        return time() - self.start_time >= self.timeout


def sleep(timeout):
    return SleepEvent(timeout)

可以看出：sleep 在調用后就會立即返回，同時一個 SleepEvent 對象會被放入消息隊列，經過timeout 秒后執行回調。

再接下來便是協程調度了：

# runner.py

from event import events_list


def run(tasks):
    for task in tasks:
        _next(task)

    while len(events_list):
        for event in events_list:
            if event.is_ready():
                events_list.remove(event)
                break


def _next(task):

    try:
        event = next(task)
        event.set_callback(lambda: _next(task)) # 1
    except StopIteration:
        pass

run 啟動了所有的子程序，并開始消息循環。每遇到一處掛起，調度器自動設置回調，并在回調中重新恢復代碼流。“1” 處巧妙地利用閉包保存狀態。

最后是主代碼：

from sleep import sleep
import runner


def task(name):
    print(name, 1)
    yield sleep(1)
    print(name, 2)
    yield sleep(2)
    print(name, 3)

if __name__ == "__main__":
    runner.run((task("hsfzxjy"), task("Jack")))

輸出：

hsfzxjy 1
Jack 1
hsfzxjy 2
Jack 2
hsfzxjy 3
Jack 3
# [Finished in 3.0s]

上面的代碼有一個不足之處，即協程函數返回的是一個 Event 對象。然而事實上只有直接操縱 IO 的協程函數才有可能接觸到這個對象。那么，對于調用了 IO 的函數的調用者，它們應該如何實現呢？

設想如下任務：

def long_add(x, y, duration=1):
    yield sleep(duration)
    return x + y


def task(duration):
    print("start:", time())
    print((yield long_add(1, 2, duration)))
    print((yield long_add(3, 4, duration)))

long_add 是 IO 的一級調用者，task 調用 long_add，并利用其返回值進行后續操作。

簡而言之，我們遇到的問題是：一個被喚起的協程函數如何喚起它的調用者？

正如在上個例子中，協程函數通過 Event 的回調與調度器交互。同理，我們也可以使用一個類似的對象，在這里我們稱其為 Future。

Future 保存在被調用者的閉包中，并由被調用者返回。而調用者通過在其上面設置回調函數，實現兩個協程函數之間的交互。

Future 的代碼如下，看起來有點像 Event：

# future.py

class Future(object):
    def __init__(self):
        super(Future, self).__init__()
        self.callback = lambda *args: None
        self._done = False

    def set_callback(self, callback):
        self.callback = callback

    def done(self, value=None):
        self._done = True
        self.callback(value)

Future 的回調函數允許接受一個參數作為返回值，以盡可能地模擬一般函數。

但這樣一來，協程函數就會有些復雜了。它們不僅要負責喚醒被調用者，還要負責與調用者之間的交互。這會產生許多重復代碼。為了 D.R.Y，我們用裝飾器封裝這一邏輯：

# co.py

from functools import wraps
from future import Future


def _next(gen, future, value=None):

    try:
        try:
            yielded_future = gen.send(value)
        except TypeError:
            yielded_future = next(gen)

        yielded_future.set_callback(lambda value: _next(gen, future, value))
    except StopIteration as e:
        future.done(e.value)


def coroutine(func):

    @wraps(func)
    def wrapper(*args, **kwargs):
        future = Future()

        gen = func(*args, **kwargs)
        _next(gen, future)
        return future

    return wrapper

被 coroutine 包裝過的生成器成為了一個普通函數，返回一個 Future 對象。_next 為喚醒的核心邏輯，通過一個類似遞歸的回調設置簡潔地實現自我喚醒。當自己執行完時，會將自己閉包內的Future對象標記為done，從而喚醒調用者。

為了適應新變化，sleep 也要做相應的更改：

from event import Event
from future import Future
from time import time


class SleepEvent(Event):

    def __init__(self, timeout):
        super(SleepEvent, self).__init__()
        self.start_time = time()
        self.timeout = timeout

    def _is_ready(self):
        return time() - self.start_time >= self.timeout


def sleep(timeout):
    future = Future()
    event = SleepEvent(timeout)
    event.set_callback(lambda: future.done())
    return future

sleep 不再返回 Event 對象，而是一致地返回 Future，并作為 Event 和 Future 之間的代理者。

基于以上更改，調度器可以更加簡潔——這是因為協程函數能夠自我喚醒：

# runner.py

from event import events_list

def run():
    while len(events_list):
        for event in events_list:
            if event.is_ready():
                events_list.remove(event)
                break

主程序：

from co import coroutine
from sleep import sleep
import runner
from time import time


@coroutine
def long_add(x, y, duration=1):
    yield sleep(duration)
    return x + y


@coroutine
def task(duration):
    print("start:", time())
    print((yield long_add(1, 2, duration)), time())
    print((yield long_add(3, 4, duration)), time())

task(2)
task(1)
runner.run()

由于我們使用了一個糟糕的事件輪詢機制，密集的計算會阻塞通往 stdout 的輸出，因而看起來所有的結果都是一起打印出來的。為此，我在打印時特地加上了時間戳，以演示協程的效果。輸出如下：

start: 1459609512.263156
start: 1459609512.263212
3 1459609513.2632613
3 1459609514.2632234
7 1459609514.263319
7 1459609516.2633028

這事實上是 tornado.gen.coroutine 的簡化版本，為了敘述方便我略去了許多細節，如異常處理以及調度優化，目的是讓大家能較清晰地了解 生成器協程 背后的機制。因此，這段代碼并不能用于實際生產中。

這，才叫精通生成器。

學習編程，不僅要知其然，亦要知其所以然。

Python 是有魔法的，只有想不到，沒有做不到。

tornado.gen.coroutine