微線程領(lǐng)域（至少在 Python 中）一直都是 Stackless Python 才能涉及的特殊增強(qiáng)部分。關(guān)于 Stackless 的話題以及最近它經(jīng)歷的變化，可能本身就值得開辟一個專欄了。但其中簡單的道理就是，在“新的 Stackless”下，延續(xù)（continuation）顯然是不合時宜的，但微線程還是這個項目 存在的理由。這一點(diǎn)很復(fù)雜……

剛開始，我們還是先來回顧一些內(nèi)容。那么，什么是微線程呢？ 微線程基本上可以說是只需要很少的內(nèi)部資源就可以運(yùn)行的進(jìn)程 ― 并且是在 Python 解釋器的單個實(shí)例中（在公共內(nèi)存空間中，等等）運(yùn)行的進(jìn)程。有了微線程，我們就可能在目前中等性能的 PC 機(jī)上運(yùn)行數(shù)以萬計的并行進(jìn)程，還可以每秒鐘幾十萬次地在上下文之間切換。對 fork() 的調(diào)用或標(biāo)準(zhǔn)的 OS 線程調(diào)用根本不能達(dá)到這個程度！甚至所謂的“輕量級”線程庫中的線程也比這里提出的微線程“重”好幾個數(shù)量級。

我在本專欄中介紹的輕便線程的含義與 OS 線程的含義有一點(diǎn)不同。就這點(diǎn)而言，它們與 Stackless 所提供的也不盡相同。在很多方面，輕便線程比大多數(shù)變體都簡單得多；大多數(shù)關(guān)于信號、鎖定及諸如此類的問題都不存在了。簡單性的代價就是，我提出了一種“協(xié)作多線程”的形式；我覺得在標(biāo)準(zhǔn) Python 框架中加入搶占并不可行（至少在非 Stackless 的 Python 2.2 中 — 沒有人知道 __future__ 會帶來什么）。

輕便線程在某種意義上會令人回想起較早的 Windows 和 MacOS 版本的協(xié)作多任務(wù)（不過是在單個應(yīng)用程序中）。然而，在另一種意義上，輕便線程只不過是在程序中表達(dá)流的另一種方式；輕便線程所做的一切（至少在原則上）都可以用“真正龐大的 if/elif 塊”技術(shù)來完成（蠻干的程序員的黔驢之計）。

一種用簡單的生成器模擬協(xié)同程序的機(jī)制。這個機(jī)制的核心部分非常簡單。 scheduler() 函數(shù)中包裝了一組生成器對象，這個函數(shù)控制將控制流委托給合適的分支的過程。這些協(xié)同程序并不是 真正的協(xié)同程序，因?yàn)樗鼈冎豢刂频?scheduler() 函數(shù)和來自該函數(shù)的分支。不過出于實(shí)用的目的，您可以用非常少的額外代碼來完成同樣的事情。 scheduler() 就是類似于下面這樣的代碼：
清單 1. 模擬協(xié)同程序的 Scheduler()

def scheduler(gendct, start):
 global cargo
 coroutine = start
 while 1:
  (coroutine, cargo) = gendct[coroutine].next()

關(guān)于這個包裝器要注意的一點(diǎn)是，每個生成器／協(xié)同程序都會生成一個包含它的預(yù)期分支目標(biāo)的元組。生成器／協(xié)同程序基本上都在 GOTO 目標(biāo)處退出。為了方便起見，我還讓生成器生成了一個標(biāo)準(zhǔn)的 cargo 容器，作為形式化在協(xié)同程序之間傳送的數(shù)據(jù)的方法 — 不過您也可以只用已經(jīng)達(dá)成一致的全局變量或回調(diào) setter/getter 函數(shù)來傳送數(shù)據(jù)。Raymond Hettinger 撰寫了一個 Python 增強(qiáng)倡議（Python Enhancement Proposal，PEP），旨在使傳送的數(shù)據(jù)能被更好地封裝；可能今后的 Python 將包括這個倡議。

新的調(diào)度程序

對于輕便線程來說，它們的需求與協(xié)同程序的需求稍有不同。不過我們還是可以在它的核心處使用 scheduler() 函數(shù)。不同之處在于，調(diào)度程序本身應(yīng)該決定分支目標(biāo)，而不是從生成器／協(xié)同程序接收分支目標(biāo)。下面讓我向您展示一個完整的測試程序和樣本：
清單 2. microthreads.py 示例腳本

from
  
   __future__ 
  
  import
  
   generators
  
  import
  
   sys, time
threads = []
TOTALSWITCHES = 
  
  10**6
NUMTHREADS = 
  
  10**5def
  
   null_factory():
 
  
  def
  
   empty():
  
  
  while1: 
  
  yield
  
   None
 
  
  return
  
   empty()
  
  def
  
   quitter():
 
  
  for
  
   n 
  
  in
  
   xrange(TOTALSWITCHES/NUMTHREADS):
  
  
  yield
  
   None
  
  def
  
   scheduler():
 
  
  global
  
   threads
 
  
  try
  
  :
  
  
  while1:
   
  
  for
  
   thread 
  
  in
  
   threads: thread.next()
 
  
  except
  
   StopIteration:
  
  
  passif
  
   __name__ == 
  
  "__main__"
  
  :
 
  
  for
  
   i 
  
  in
  
   range(NUMTHREADS):
  threads.append(null_factory())
 threads.append(quitter())
 starttime = time.clock()
 scheduler()
 
  
  print"TOTAL TIME: "
  
  , time.clock()-starttime
 
  
  print"TOTAL SWITCHES:"
  
  , TOTALSWITCHES
 
  
  print"TOTAL THREADS: "
  
  , NUMTHREADS

這大概就是您能夠選擇的最簡單的輕便線程調(diào)度程序了。每個線程都按固定順序進(jìn)入，而且每個線程都有同樣的優(yōu)先級。接下來，讓我們來看看如何處理細(xì)節(jié)問題。和前面部分所講的協(xié)同程序一樣，編寫輕便線程時應(yīng)該遵守一些約定。

處理細(xì)節(jié)

大多數(shù)情況下，輕便線程的生成器都應(yīng)該包括在 while 1: 循環(huán)中。這里設(shè)置調(diào)度程序的方法將導(dǎo)致在其中一個線程停止時整個調(diào)度程序停止。這在某種意義上“健壯性”不如 OS 線程 ― 不過在 scheduler() 的循環(huán) 內(nèi)捕獲異常不會比在循環(huán)外需要更多的機(jī)器資源。而且，我們可以從 threads 列表刪除線程，而不必終止（由它本身或其它線程終止）。我們其實(shí)并沒有提供讓刪除更加容易的詳細(xì)方法；不過比較常用的擴(kuò)展方法可能是將線程存儲在字典或某種其它的結(jié)構(gòu)中，而不是列表中。

該示例說明了最后終止調(diào)度程序循環(huán)的一種合理的方法。 quitter() 是一種特殊的生成器／線程，它監(jiān)視某種條件（在本示例中只是一個上下文切換的計數(shù)），并在條件滿足時拋出 StopIteration （本示例中不捕獲其它異常）。請注意，在終止之后，其它所有生成器還是完整的，如果需要，還可以在今后恢復(fù)（在微線程調(diào)度程序或其它程序中）。顯然，如果需要，您可以 delete 這些生成器／線程。

這里討論的示例使用了特殊的無意義線程。它們什么也不做，而且以一種可能性最小的形式實(shí)現(xiàn)這一點(diǎn)。我們這樣建立該示例是為了說明一點(diǎn) ― 輕便線程的內(nèi)在開銷是非常低的。在一臺比較老的只有 64 MB 內(nèi)存的 Windows 98 Pentium II 膝上型電腦上創(chuàng)建 100,000 個輕便線程是輕而易舉的（如果達(dá)到了一百萬個線程，就會出現(xiàn)長時間的磁盤“猛轉(zhuǎn)”）。請用 OS 線程試試看！ 而且，在這個比較慢的 366 MHz 芯片上可以在大約 10 秒內(nèi)執(zhí)行一百萬次上下文切換（所涉及的線程數(shù)對耗時并無重大影響）。顯然，真正的輕便線程應(yīng)該 做一些事情，而這將根據(jù)任務(wù)使用更多的資源。不過線程本身卻贏得了“輕便”的名聲。

切換開銷

在輕便線程之間切換開銷很小，但還不是完全沒有開銷。為了測試這種情況，我構(gòu)建了一個執(zhí)行 某種工作（不過大約是您在線程中按道理可以完成的最少量）的示例。因?yàn)榫€程調(diào)度程序 真的等同于“執(zhí)行 A，接著執(zhí)行 B，然后執(zhí)行 C，等等”的指令，所以要在主函數(shù)中創(chuàng)建一個完全并行的情況也不困難。
清單 3. overhead.py 示例腳本

from
  
   __future__ 
  
  import
  
   generators
  
  import
  
   time
TIMES = 100000
  
  def
  
   stringops():
 
  
  for
  
   n 
  
  in
  
   xrange(TIMES):
  s = 
  
  "Mary had a little lamb"
  
  
  s = s.upper()
  s = 
  
  "Mary had a little lamb"
  
  
  s = s.lower()
  s = 
  
  "Mary had a little lamb"
  
  
  s = s.replace('a','A')
  
  def
  
   scheduler():
 
  
  for
  
   n 
  
  in
  
   xrange(TIMES):
  
  
  for
  
   thread 
  
  in
  
   threads: thread.next()
  
  def
  
   upper():
 
  
  while1:
  s = 
  
  "Mary had a little lamb"
  
  
  s = s.upper()
  
  
  yield
  
   None
  
  def
  
   lower():
 
  
  while1:
  s = 
  
  "Mary had a little lamb"
  
  
  s = s.lower()
  
  
  yield
  
   None
  
  def
  
   replace():
 
  
  while1:
  s = 
  
  "Mary had a little lamb"
  
  
  s = s.replace(
  
  'a'
  
  ,
  
  'A'
  
  )
  
  
  yield
  
   None
  
  if
  
   __name__==
  
  '__main__':
  
  
 start = time.clock()
 stringops()
 looptime = time.clock()-start
 
  
  print"LOOP TIME:"
  
  , looptime
 
  
  global
  
   threads
 threads.append(upper())
 threads.append(lower())
 threads.append(replace())
 start = time.clock()
 scheduler()
 threadtime = time.clock()-start
 
  
  print"THREAD TIME:"
  
  , threadtime

結(jié)果表明，在直接循環(huán)的版本運(yùn)行一次的時間內(nèi)，輕便線程的版本運(yùn)行了兩次還多一點(diǎn)點(diǎn) ― 也就相當(dāng)于在上面提到的機(jī)器上，輕便線程運(yùn)行了不到 3 秒，而直接循環(huán)運(yùn)行了超過 6 秒。顯然，如果每個工作單元都相當(dāng)于單個字符串方法調(diào)用的兩倍、十倍或一百倍，那么所花費(fèi)的線程開銷比例就相應(yīng)地更小了。

設(shè)計線程

輕便線程可以（而且通常應(yīng)該）比單獨(dú)的概念性操作規(guī)模更大。無論是何種線程，都是用來表示描述一個特定 任務(wù)或 活動所需的流上下文的量。但是，任務(wù)花費(fèi)的時間／大小可能比我們希望在單獨(dú)線程上下文中使用的要多／大。搶占將自動處理這種問題，不需要應(yīng)用程序開發(fā)者作出任何特定干涉。不幸的是，輕便線程用戶需要注意“好好地處理”其它線程。

至少，輕便線程應(yīng)該設(shè)計得足夠周全，在完成概念性操作時應(yīng)該能夠 yield 。調(diào)度程序?qū)⒒氐竭@里以進(jìn)行下一步。舉例來說：
清單 4. 偽碼友好的輕便線程

def nicethread():
    while 1:
        ...operation A...
        yield None
        ...operation B...
        yield None
        ...operation C...
        yield None

多數(shù)情況下，好的設(shè)計將比在基本操作之間的邊界 yield 更多次。雖然如此，通常在概念上“基本”的東西都涉及對一個大集合的循環(huán)。如果情況如此（根據(jù)循環(huán)體耗費(fèi)時間的程度），在循環(huán)體中加入一到兩個 yield （可能在特定數(shù)量的循環(huán)迭代執(zhí)行過后再次發(fā)生）可能會有所幫助。和優(yōu)先權(quán)線程的情況不同，一個行為不良的輕便線程會獲取無限量的獨(dú)占處理器時間。

調(diào)度的其它部分

迄今為止，上面的示例只展示了形式最基本的幾個線程調(diào)度程序?？赡軐?shí)現(xiàn)的還有很多（這個問題與設(shè)計一個好的生成器／線程沒什么關(guān)系）。讓我來順便向您展示幾個傳送中可能出現(xiàn)的增強(qiáng)。
更好的線程管理

一個簡單的 threads 列表就可以使添加調(diào)度程序要處理的生成器／線程非常容易。但是這種數(shù)據(jù)結(jié)構(gòu)并不能使刪除或暫掛不再相關(guān)的線程變得容易。字典或類可能是線程管理中更好的數(shù)據(jù)結(jié)構(gòu)。下面是一個快捷的示例，這個類能夠（幾乎能夠）順便訪問示例中的 threads 列表：
清單 5. 線程管理的 Python 類示例

class
    
     ThreadPool: 
  
    
    """Enhanced threads list as class
  threads = ThreadPool()
  threads.append(threadfunc) # not generator object
  if threads.query(num) <<has some property>>:
    threads.remove(num)
  """def
    
     __init__(self):
    self.threadlist = []
    self.threaddict = {}
    self.avail = 
    
    1def
    
     __getitem__(self, n):
    
    
    return
    
     self.threadlist[n]
  
    
    def
    
     append(self, threadfunc, docstring=None):
    
    
    # Argument is the generator func, not the gen object
# Every threadfunc should contain a docstring
    
    
    docstring = docstring 
    
    or
    
     threadfunc.__doc__
    self.threaddict[self.avail] = (docstring, threadfunc())
    self.avail += 
    
    1
    self.threadlist = [p[
    
    1] 
    
    for
    
     p 
    
    in
    
     self.threaddict.values()]
    
    
    return
    
     self.avail-
    
    1# return the threadIDdef
    
     remove(self, threadID):
    
    
    del
    
     self.threaddict[threadID]
    self.threadlist = [p[
    
    1] 
    
    for
    
     p 
    
    in
    
     self.threaddict.values()]
  
    
    def
    
     query(self, threadID):
    
    
    "
    
    Information on thread, 
    
    if
    
     it exists (otherwise None)
    
    
    return
    
     self.threaddict.get(threadID,[None])[0]

您可以實(shí)現(xiàn)更多內(nèi)容，而這是個好的起點(diǎn)。
線程優(yōu)先級

在簡單的示例中，所有線程都獲得調(diào)度程序同等的關(guān)注。至少有兩種普通方法可以實(shí)現(xiàn)調(diào)優(yōu)程度更好的線程優(yōu)先級系統(tǒng)。一個優(yōu)先級系統(tǒng)可以只對“高優(yōu)先級”線程投入比低優(yōu)先級線程更多的注意力。我們可以用一種直接的方式實(shí)現(xiàn)它，就是創(chuàng)建一個新類 PriorityThreadPool(ThreadPool) ，這個類在線程迭代期間更頻繁地返回更重要的線程。最簡單的方法可能會在 .__getitem__() 方法中連續(xù)多次返回某些線程。那么，高優(yōu)先級線程就可能接收到兩個，或多個，或一百個連續(xù)的“時間片”，而不只是原來的一個。這里的一個（非常弱的）“實(shí)時”變量最多可能返回散落在線程列表中各處的重要線程的多個副本。這將增加服務(wù)于高優(yōu)先級線程的實(shí)際頻率，而不只是它們受到的所有關(guān)注。

在純 Python 中使用更復(fù)雜的線程優(yōu)先級方法可能不是很容易（不過它是使用某種第三方特定于 OS／處理器的庫來實(shí)現(xiàn)的）。調(diào)度程序不是只給高優(yōu)先級線程一個時間片的整型數(shù)，它還可以測量每個輕便線程中實(shí)際花費(fèi)的時間，然后動態(tài)調(diào)整線程調(diào)度，使其對等待處理的線程更加“公平”（也許公平性和線程優(yōu)先級是相關(guān)的）。不幸的是，Python 的 time.clock() 和它的系列都不是精度足夠高的計時器，不足以使這種方式有效。另一方面，沒有什么可以阻止“多時間片”方法中處理不足的線程去動態(tài)提高它自己的優(yōu)先級。
將微線程和協(xié)作程序結(jié)合在一起

為了創(chuàng)建一個輕便線程（微線程）調(diào)度程序，我刪除了協(xié)作程序邏輯“please branch to here”。這樣做其實(shí)并不必要。示例中的輕便線程生成的通常都是 None ，而不是跳轉(zhuǎn)目標(biāo)。我們完全可以把這兩個概念結(jié)合在一起：如果協(xié)同程序／線程生成了跳轉(zhuǎn)目標(biāo)，調(diào)度程序就可以跳轉(zhuǎn)到被請求的地方（也許，除非被線程優(yōu)先級覆蓋）。然而，如果協(xié)同程序／線程只生成 None ，調(diào)度程序就可以自己決定下一步要關(guān)注哪個適當(dāng)?shù)木€程。決定（以及編寫）一個任意的跳轉(zhuǎn)究竟會如何與線性線程隊列交互將涉及到不少工作，不過這些工作中沒有什么特別神秘的地方。

快速而廉價 — 為什么不喜歡它呢？

微線程模式（或者“輕便線程”）基本上可以歸結(jié)為 Python 中流控制的另一種奇怪的風(fēng)格。本專欄的前面幾個部分已經(jīng)談到了另外幾種風(fēng)格。有各種控制機(jī)制的引人之處在于，它讓開發(fā)者將代碼功能性隔離在其邏輯組件內(nèi)，并最大化代碼的上下文相關(guān)性。

其實(shí)，要實(shí)現(xiàn)做任何可能做到的事的 可能性并不復(fù)雜（只要用一個“l(fā)oop”和一個“if”就可以了）。對于輕易地分解為很多細(xì)小的“代理”、“服務(wù)器”或“進(jìn)程”的一類問題來說，輕便線程可能是表達(dá)應(yīng)用程序的底層“業(yè)務(wù)邏輯”的最清楚的模型。當(dāng)然，輕便線程與一些大家更熟知的流機(jī)制相比速度可能非?？?，就這點(diǎn)而言并無大礙。

最新評論