摘要：徘徊和行程所用的時(shí)間使用指數(shù)分布生成，我們將時(shí)間設(shè)為分鐘數(shù)，以便顯示清楚。迭代表示各輛出租車的進(jìn)程在各輛出租車上調(diào)用函數(shù)，預(yù)激協(xié)程。

前兩篇我們已經(jīng)介紹了python 協(xié)程的使用和yield from 的原理，這一篇，我們用一個(gè)例子來(lái)揭示如何使用協(xié)程在單線程中管理并發(fā)活動(dòng)。。

Wiki上的定義是：

離散事件仿真將系統(tǒng)隨時(shí)間的變化抽象成一系列的離散時(shí)間點(diǎn)上的事件，通過(guò)按照事件時(shí)間順序處理事件來(lái)演進(jìn)，是一種事件驅(qū)動(dòng)的仿真世界觀。離散事件仿真將系統(tǒng)的變化看做一個(gè)事件，因此系統(tǒng)任何的變化都只能是通過(guò)處理相應(yīng)的事件來(lái)實(shí)現(xiàn)，在兩個(gè)相鄰的事件之間，系統(tǒng)狀態(tài)維持前一個(gè)事件發(fā)生后的狀態(tài)不變。

人話說(shuō)就是一種把系統(tǒng)建模成一系列事件的仿真系統(tǒng)。在離散事件仿真中，仿真“鐘”向前推進(jìn)的量不是固定的，而是直接推進(jìn)到下一個(gè)事件模型的模擬時(shí)間。

假設(shè)我們抽象模擬出租車的運(yùn)營(yíng)過(guò)程，其中一個(gè)事件是乘客上車，下一個(gè)事件則是乘客下車。不管乘客做了5分鐘還是50分鐘，一旦下車，仿真鐘就會(huì)更新，指向此次運(yùn)營(yíng)的結(jié)束時(shí)間。

事件？是不是想到了協(xié)程！

協(xié)程恰好為實(shí)現(xiàn)離散事件仿真提供了合理的抽象。

第一門面向?qū)ο蟮恼Z(yǔ)音 Simula 引入?yún)f(xié)程這個(gè)概念就是為了支持仿真。
Simpy 是一個(gè)實(shí)現(xiàn)離散事件仿真的Python包，通過(guò)一個(gè)協(xié)程表示離散事件仿真系統(tǒng)的各個(gè)進(jìn)程。

仿真程序會(huì)創(chuàng)建幾輛出租車，每輛出租車會(huì)拉幾個(gè)乘客，然后回家。出租車會(huì)首先駛離車庫(kù)，四處徘徊，尋找乘客；拉到乘客后，行程開(kāi)始；乘客下車后，繼續(xù)四處徘徊。

徘徊和行程所用的時(shí)間使用指數(shù)分布生成，我們將時(shí)間設(shè)為分鐘數(shù)，以便顯示清楚。

完整代碼如下：(taxi_sim.py)

#! -*- coding: utf-8 -*-
import random
import collections
import queue
import argparse

DEFAULT_NUMBER_OF_TAXIS = 3
DEFAULT_END_TIME = 180
SEARCH_DURATION = 5
TRIP_DURATION = 20
DEPARTURE_INTERAVAL = 5


# time 是事件發(fā)生的仿真時(shí)間，proc 是出租車進(jìn)程實(shí)例的編號(hào)，action是描述活動(dòng)的字符串
Event = collections.namedtuple("Event", "time proc action")

# 開(kāi)始 出租車進(jìn)程
# 每輛出租車調(diào)用一次taxi_process 函數(shù)，創(chuàng)建一個(gè)生成器對(duì)象，表示各輛出租車的運(yùn)營(yíng)過(guò)程。
def taxi_process(ident, trips, start_time=0):
    """
    每次狀態(tài)變化時(shí)向創(chuàng)建事件，把控制權(quán)交給仿真器
    :param ident: 出租車編號(hào)
    :param trips: 出租車回家前的行程數(shù)量
    :param start_time: 離開(kāi)車庫(kù)的時(shí)間
    :return: 
    """
    time = yield Event(start_time, ident, "leave garage") # 產(chǎn)出的第一個(gè)Event
    for i in range(trips):  # 每次行程都會(huì)執(zhí)行一遍這個(gè)代碼塊
        # 產(chǎn)出一個(gè)Event實(shí)例，表示拉到了乘客 協(xié)程在這里暫停 等待下一次send() 激活
        time = yield Event(time, ident, "pick up passenger")
         # 產(chǎn)出一個(gè)Event實(shí)例，表示乘客下車 協(xié)程在這里暫停 等待下一次send() 激活
        time = yield Event(time, ident, "drop off passenger")
    # 指定的行程數(shù)量完成后，for 循環(huán)結(jié)束，最后產(chǎn)出 "going home" 事件。協(xié)程最后一次暫停
    yield Event(time, ident, "going home")
    # 協(xié)程執(zhí)行到最后 拋出StopIteration 異常


def compute_duration(previous_action):
    """使用指數(shù)分布計(jì)算操作的耗時(shí)"""
    if previous_action in ["leave garage", "drop off passenger"]:
        # 新?tīng)顟B(tài)是四處徘徊
        interval = SEARCH_DURATION
    elif previous_action == "pick up passenger":
        # 新?tīng)顟B(tài)是開(kāi)始行程
        interval = TRIP_DURATION
    elif previous_action == "going home":
        interval = 1
    else:
        raise ValueError("Unkonw previous_action: %s" % previous_action)
    return int(random.expovariate(1/interval)) + 1


# 開(kāi)始仿真
class Simulator:

    def __init__(self, procs_map):
        self.events = queue.PriorityQueue()  # 帶優(yōu)先級(jí)的隊(duì)列 會(huì)按時(shí)間正向排序
        self.procs = dict(procs_map) # 從獲取的procs_map 參數(shù)中創(chuàng)建本地副本，為了不修改用戶傳入的值

    def run(self, end_time):
        """
        調(diào)度并顯示事件，直到時(shí)間結(jié)束
        :param end_time:  結(jié)束時(shí)間 只需要指定一個(gè)參數(shù)
        :return: 
        """
        # 調(diào)度各輛出租車的第一個(gè)事件
        for iden, proc in sorted(self.procs.items()):
            first_event = next(proc)  # 預(yù)激協(xié)程 并產(chǎn)出一個(gè) Event 對(duì)象
            self.events.put(first_event)  # 把各個(gè)事件加到self.events 屬性表示的 PriorityQueue對(duì)象中

        # 此次仿真的主循環(huán)
        sim_time = 0  # 把 sim_time 歸0
        while sim_time < end_time:
            if self.events.empty(): # 事件全部完成后退出循環(huán)
                print("*** end of event ***")
                break
            current_event = self.events.get() # 獲取優(yōu)先級(jí)最高(time 屬性最小)的事件
            sim_time, proc_id, previous_action = current_event # 更新 sim_time
            print("taxi:", proc_id, proc_id * "  ", current_event)
            active_proc = self.procs[proc_id]  # 從self.procs 字典中獲取表示當(dāng)前活動(dòng)的出租車協(xié)程
            next_time = sim_time + compute_duration(previous_action)
            try:
                next_event = active_proc.send(next_time)  # 把計(jì)算得到的時(shí)間發(fā)送給出租車協(xié)程。協(xié)程會(huì)產(chǎn)出下一個(gè)事件，或者拋出 StopIteration
            except StopIteration:
                del self.procs[proc_id]  # 如果有異常 表示已經(jīng)退出， 刪除這個(gè)協(xié)程
            else:
                self.events.put(next_event)  # 如果沒(méi)有異常，把next_event 加入到隊(duì)列
        else:  # 如果超時(shí) 則走到這里
            msg = "*** end of simulation time: {} event pendding ***"
            print(msg.format(self.events.qsize()))



def main(end_time=DEFAULT_END_TIME, num_taxis=DEFAULT_NUMBER_OF_TAXIS,
         seed=None):
    """初始化隨機(jī)生成器，構(gòu)建過(guò)程，運(yùn)行仿真程序"""
    if seed is not None:
        random.seed(seed)  # 獲取可復(fù)現(xiàn)的結(jié)果
    # 構(gòu)建taxis 字典。值是三個(gè)參數(shù)不同的生成器對(duì)象。
    taxis = {i: taxi_process(i, (i + 1) * 2, i*DEPARTURE_INTERAVAL)
             for i in range(num_taxis)}
    sim = Simulator(taxis)
    sim.run(end_time)


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Taxi fleet simulator.")
    parser.add_argument("-e", "--end-time", type=int,
                        default=DEFAULT_END_TIME,
                        help="simulation end time; default=%s" % DEFAULT_END_TIME)
    parser.add_argument("-t", "--taxis", type=int,
                        default=DEFAULT_NUMBER_OF_TAXIS,
                        help="number of taxis running; default = %s" % DEFAULT_NUMBER_OF_TAXIS)
    parser.add_argument("-s", "--seed", type=int, default=None,
                        help="random generator seed (for testing)")

    args = parser.parse_args()
    main(args.end_time, args.taxis, args.seed)

運(yùn)行程序，

# -s 3 參數(shù)設(shè)置隨機(jī)生成器的種子，以便調(diào)試的時(shí)候隨機(jī)數(shù)不變，輸出相同的結(jié)果
python taxi_sim.py -s 3

輸出結(jié)果如下圖

從結(jié)果我們可以看出，3輛出租車的行程是交叉進(jìn)行的。不同顏色的箭頭代表不同出租車從乘客上車到乘客下車的跨度。

從結(jié)果可以看出：

出租車每5隔分鐘從車庫(kù)出發(fā)

0 號(hào)出租車2分鐘后拉到乘客（time=2），1號(hào)出租車3分鐘后拉到乘客（time=8），2號(hào)出租車5分鐘后拉到乘客（time=15）

0 號(hào)出租車?yán)藘蓚€(gè)乘客

1 號(hào)出租車?yán)?個(gè)乘客

2 號(hào)出租車?yán)?個(gè)乘客

在此次示中，所有排定的事件都在默認(rèn)的仿真時(shí)間內(nèi)完成

我們先在控制臺(tái)中調(diào)用taxi_process 函數(shù)，自己駕駛一輛出租車，示例如下：

In [1]: from taxi_sim import taxi_process
# 創(chuàng)建一個(gè)生成器，表示一輛出租車 編號(hào)是13 從t=0 開(kāi)始，有兩次行程
In [2]: taxi = taxi_process(ident=13, trips=2, start_time=0) 

In [3]: next(taxi) # 預(yù)激協(xié)程
Out[3]: Event(time=0, proc=13, action="leave garage")

# 發(fā)送當(dāng)前時(shí)間 在控制臺(tái)中，變量_綁定的是前一個(gè)結(jié)果
# _.time + 7 是 0 + 7
In [4]: taxi.send(_.time+7) 
Out[4]: Event(time=7, proc=13, action="pick up passenger") # 這個(gè)事件有for循環(huán)在第一個(gè)行程的開(kāi)頭產(chǎn)出

# 發(fā)送_.time+12 表示這個(gè)乘客用時(shí)12分鐘
In [5]: taxi.send(_.time+12)
Out[5]: Event(time=19, proc=13, action="drop off passenger")

# 徘徊了29 分鐘
In [6]: taxi.send(_.time+29)
Out[6]: Event(time=48, proc=13, action="pick up passenger")

# 乘坐了50分鐘
In [7]: taxi.send(_.time+50)
Out[7]: Event(time=98, proc=13, action="drop off passenger")

# 兩次行程結(jié)束 for 循環(huán)結(jié)束產(chǎn)出"going home"
In [8]: taxi.send(_.time+5)
Out[8]: Event(time=103, proc=13, action="going home")

# 再發(fā)送值，會(huì)執(zhí)行到末尾 協(xié)程返回后 拋出 StopIteration 異常
In [9]: taxi.send(_.time+10)
---------------------------------------------------------------------------
StopIteration                            Traceback (most recent call last)
 in ()
----> 1 taxi.send(_.time+10)

StopIteration:

在這個(gè)示例中，我們用控制臺(tái)模擬仿真主循環(huán)。從taxi協(xié)程中產(chǎn)出的Event實(shí)例中獲取 .time 屬性，隨意加一個(gè)數(shù)，然后調(diào)用send()方法發(fā)送兩數(shù)之和，重新激活協(xié)程。

在taxi_sim.py 代碼中，出租車協(xié)程由 Simulator.run 方法中的主循環(huán)驅(qū)動(dòng)。

Simulator 類的主要數(shù)據(jù)結(jié)構(gòu)如下：

self.events

PriorityQueue 對(duì)象，保存Event實(shí)例。元素可以放進(jìn)PriorityQueue對(duì)象中，然后按 item[0](對(duì)象的time 屬性)依序取出（按從小到大）。

self.procs

一個(gè)字典，把出租車的編號(hào)映射到仿真過(guò)程的進(jìn)程（表示出租車生成器的對(duì)象）。這個(gè)屬性會(huì)綁定前面所示的taxis字典副本。

優(yōu)先隊(duì)列是離散事件仿真系統(tǒng)的基礎(chǔ)構(gòu)件：創(chuàng)建事件的順序不定，放入這種隊(duì)列后，可以按各個(gè)事件排定的順序取出。

比如，我們把兩個(gè)事件放入隊(duì)列：

Event(time=14, proc=0, action="pick up passenger")
Event(time=10, proc=1, action="pick up passenger")

這個(gè)意思是 0號(hào)出租車14分拉到一個(gè)乘客，1號(hào)出租車10分拉到一個(gè)乘客。但是主循環(huán)獲取的第一個(gè)事件將是

Event(time=10, proc=1, action="pick up passenger")

下面我們分析一下仿真系統(tǒng)的主算法--Simulator.run 方法。

迭代表示各輛出租車的進(jìn)程

在各輛出租車上調(diào)用next()函數(shù)，預(yù)激協(xié)程。

把各個(gè)事件放入Simulator類的self.events屬性中。

滿足 sim_time < end_time 條件是，運(yùn)行仿真系統(tǒng)的主循環(huán)。

檢查self.events 屬性是否為空；如果為空，跳出循環(huán)

從self.events 中獲取當(dāng)前事件

顯示獲取的Event對(duì)象

獲取curent_event 的time 屬性，更新仿真時(shí)間

把時(shí)間發(fā)送給current_event 的pro屬性標(biāo)識(shí)的協(xié)程，產(chǎn)出下一個(gè)事件

把next_event 添加到self.events 隊(duì)列中，排定 next_event

我們代碼中 while 循環(huán)有一個(gè)else 語(yǔ)句，仿真系統(tǒng)到達(dá)結(jié)束時(shí)間后，代碼會(huì)執(zhí)行else中的語(yǔ)句。

這個(gè)示例主要是想說(shuō)明如何在一個(gè)主循環(huán)中處理事件，以及如何通過(guò)發(fā)送數(shù)據(jù)驅(qū)動(dòng)協(xié)程，同時(shí)解釋了如何使用生成器代替線程和回調(diào)，實(shí)現(xiàn)并發(fā)。

并發(fā)： 多個(gè)任務(wù)交替執(zhí)行
并行： 多個(gè)任務(wù)同時(shí)執(zhí)行

到這里 Python協(xié)程系列的三篇文章就結(jié)束了。

我們會(huì)看到，協(xié)程做面向事件編程時(shí)，會(huì)不斷把控制權(quán)讓步給主循環(huán)，激活并向前運(yùn)行其他協(xié)程，從而執(zhí)行各個(gè)并發(fā)活動(dòng)。
協(xié)程一種協(xié)作式多任務(wù)：協(xié)程顯式自主的把控制權(quán)讓步給中央調(diào)度程序。
多線程實(shí)現(xiàn)的是搶占式多任務(wù)。調(diào)度程序可以在任何時(shí)刻暫停線程，把控制權(quán)交給其他線程

python 協(xié)程1：協(xié)程10分鐘入門

python 協(xié)程2：yield from 從入門到精通

再次說(shuō)明一下，這幾篇是《流暢的python》一書(shū)的讀書(shū)筆記，作者提供了大量的擴(kuò)展閱讀，有興趣的可以看一下。

Generator Tricks for Systems Programmers

A Curious Course on Coroutines and Concurrency

Generators: The Final Frontier

greedy algorithm with coroutines

BinaryTree類、一個(gè)簡(jiǎn)單的XML解析器、和一個(gè)任務(wù)調(diào)度器Proposal for a yield from statement for Python

考慮用協(xié)程操作多個(gè)函數(shù)

最后，感謝女朋友支持。