引言

解釋器環(huán)境：python3.5.1

我們都知道python網(wǎng)絡(luò)編程的兩大必學(xué)模塊socket和socketserver，其中的socketserver是一個(gè)支持IO多路復(fù)用和多線程、多進(jìn)程的模塊。一般我們?cè)趕ocketserver服務(wù)端代碼中都會(huì)寫(xiě)這么一句：

server = socketserver.ThreadingTCPServer(settings.IP_PORT, MyServer)

ThreadingTCPServer這個(gè)類(lèi)是一個(gè)支持多線程和TCP協(xié)議的socketserver，它的繼承關(guān)系是這樣的：
class ThreadingTCPServer(ThreadingMixIn, TCPServer): pass

右邊的TCPServer實(shí)際上是它主要的功能父類(lèi)，而左邊的ThreadingMixIn則是實(shí)現(xiàn)了多線程的類(lèi)，它自己本身則沒(méi)有任何代碼。
MixIn在python的類(lèi)命名中，很常見(jiàn)，一般被稱(chēng)為“混入”，戲稱(chēng)“亂入”，通常為了某種重要功能被子類(lèi)繼承。

class ThreadingMixIn:
  
 daemon_threads = False
 
 def process_request_thread(self, request, client_address):  
  try:
   self.finish_request(request, client_address)
   self.shutdown_request(request)
  except:
   self.handle_error(request, client_address)
   self.shutdown_request(request)
 
 def process_request(self, request, client_address):
   
  t = threading.Thread(target = self.process_request_thread,
        args = (request, client_address))
  t.daemon = self.daemon_threads
  t.start()

在ThreadingMixIn類(lèi)中，其實(shí)就定義了一個(gè)屬性，兩個(gè)方法。在process_request方法中實(shí)際調(diào)用的正是python內(nèi)置的多線程模塊threading。這個(gè)模塊是python中所有多線程的基礎(chǔ)，socketserver本質(zhì)上也是利用了這個(gè)模塊。

一、線程

以上那一段，可以不用看！舉個(gè)例子，廠家要生產(chǎn)某個(gè)產(chǎn)品，在它的生產(chǎn)基地建設(shè)了很多廠房，每個(gè)廠房?jī)?nèi)又有多條流水生產(chǎn)線。所有廠房配合將整個(gè)產(chǎn)品生產(chǎn)出來(lái)，某個(gè)廠房?jī)?nèi)的所有流水線將這個(gè)廠房負(fù)責(zé)的產(chǎn)品部分生產(chǎn)出來(lái)。每個(gè)廠房擁有自己的材料庫(kù)，廠房?jī)?nèi)的生產(chǎn)線共享這些材料。而每一個(gè)廠家要實(shí)現(xiàn)生產(chǎn)必須擁有至少一個(gè)廠房一條生產(chǎn)線。那么這個(gè)廠家就是某個(gè)應(yīng)用程序；每個(gè)廠房就是一個(gè)進(jìn)程；每條生產(chǎn)線都是一個(gè)線程。

1.1 普通的多線程

在python中，threading模塊提供線程的功能。通過(guò)它，我們可以輕易的在進(jìn)程中創(chuàng)建多個(gè)線程。下面是個(gè)例子：

import threading
import time
 
def show(arg):
 time.sleep(1)
 print('thread'+str(arg))
 
for i in range(10):
 t = threading.Thread(target=show, args=(i,))
 t.start()
 
print('main thread stop')

上述代碼創(chuàng)建了10個(gè)“前臺(tái)”線程，然后控制器就交給了CPU，CPU根據(jù)指定算法進(jìn)行調(diào)度，分片執(zhí)行指令。

下面是Thread類(lèi)的主要方法：

start 線程準(zhǔn)備就緒，等待CPU調(diào)度
setName 為線程設(shè)置名稱(chēng)
getName 獲取線程名稱(chēng)
setDaemon 設(shè)置為后臺(tái)線程或前臺(tái)線程（默認(rèn)）
如果是后臺(tái)線程，主線程執(zhí)行過(guò)程中，后臺(tái)線程也在進(jìn)行，主線程執(zhí)行完畢后，后臺(tái)線程不論成功與否，均停止。如果是前臺(tái)線程，主線程執(zhí)行過(guò)程中，前臺(tái)線程也在進(jìn)行，主線程執(zhí)行完畢后，等待前臺(tái)線程也執(zhí)行完成后，程序停止。
join 逐個(gè)執(zhí)行每個(gè)線程，執(zhí)行完畢后繼續(xù)往下執(zhí)行，該方法使得多線程變得無(wú)意義。
run 線程被cpu調(diào)度后自動(dòng)執(zhí)行線程對(duì)象的run方法　　

1.2 自定義線程類(lèi)

對(duì)于threading模塊中的Thread類(lèi)，本質(zhì)上是執(zhí)行了它的run方法。因此可以自定義線程類(lèi)，讓它繼承Thread類(lèi)，然后重寫(xiě)run方法。

import threading
class MyThreading(threading.Thread):

 def __init__(self,func,arg):
  super(MyThreading,self).__init__()
  self.func = func
  self.arg = arg

 def run(self):
  self.func(self.arg)

def f1(args):
 print(args)

obj = MyThreading(f1, 123)
obj.start()

1.3 線程鎖

CPU執(zhí)行任務(wù)時(shí)，在線程之間是進(jìn)行隨機(jī)調(diào)度的，并且每個(gè)線程可能只執(zhí)行n條代碼后就轉(zhuǎn)而執(zhí)行另外一條線程。由于在一個(gè)進(jìn)程中的多個(gè)線程之間是共享資源和數(shù)據(jù)的，這就容易造成資源搶奪或臟數(shù)據(jù)，于是就有了鎖的概念，限制某一時(shí)刻只有一個(gè)線程能訪問(wèn)某個(gè)指定的數(shù)據(jù)。

1.3.1 未使用鎖

#!/usr/bin/env python
# -*- coding:utf-8 -*-
import threading
import time
NUM = 0
def show():
 global NUM
 NUM += 1
 name = t.getName()
 time.sleep(1)  # 注意，這行語(yǔ)句的位置很重要，必須在NUM被修改后，否則觀察不到臟數(shù)據(jù)的現(xiàn)象。
 print(name, "執(zhí)行完畢后，NUM的值為： ", NUM)

for i in range(10):
 t = threading.Thread(target=show)
 t.start()
print('main thread stop')

上述代碼運(yùn)行后，結(jié)果如下：

main thread stop
Thread-1 執(zhí)行完畢后，NUM的值為： 10
Thread-2 執(zhí)行完畢后，NUM的值為： 10
Thread-4 執(zhí)行完畢后，NUM的值為： 10
Thread-9 執(zhí)行完畢后，NUM的值為： 10
Thread-3 執(zhí)行完畢后，NUM的值為： 10
Thread-6 執(zhí)行完畢后，NUM的值為： 10
Thread-8 執(zhí)行完畢后，NUM的值為： 10
Thread-7 執(zhí)行完畢后，NUM的值為： 10
Thread-5 執(zhí)行完畢后，NUM的值為： 10
Thread-10 執(zhí)行完畢后，NUM的值為： 10

由此可見(jiàn)，由于線程同時(shí)訪問(wèn)一個(gè)數(shù)據(jù)，產(chǎn)生了錯(cuò)誤的結(jié)果。為了解決這個(gè)問(wèn)題，python在threading模塊中定義了幾種線程鎖類(lèi)，分別是：

1. Lock 普通鎖（不可嵌套）
2. RLock 普通鎖（可嵌套）常用
3. Semaphore 信號(hào)量
4. event 事件
5. condition 條件
   

1.3.2 普通鎖Lock和RLock

類(lèi)名：Lock或RLock

普通鎖，也叫互斥鎖，是獨(dú)占的，同一時(shí)刻只有一個(gè)線程被放行。

import time
import threading

NUM = 10
def func(lock):
 global NUM
 lock.acquire() # 讓鎖開(kāi)始起作用
 NUM -= 1
 time.sleep(1)
 print(NUM)
 lock.release() # 釋放鎖
 
lock = threading.Lock() # 實(shí)例化一個(gè)鎖對(duì)象
for i in range(10):
 t = threading.Thread(target=func, args=(lock,)) # 記得把鎖當(dāng)作參數(shù)傳遞給func參數(shù)
 t.start()

以上是threading模塊的Lock類(lèi)，它不支持嵌套鎖。RLcok類(lèi)的用法和Lock一模一樣，但它支持嵌套，因此我們一般直接使用RLcok類(lèi)。

1.3.3 信號(hào)量(Semaphore)

類(lèi)名：BoundedSemaphore

這種鎖允許一定數(shù)量的線程同時(shí)更改數(shù)據(jù)，它不是互斥鎖。比如地鐵安檢，排隊(duì)人很多，工作人員只允許一定數(shù)量的人進(jìn)入安檢區(qū)，其它的人繼續(xù)排隊(duì)。

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import time
import threading

def run(n):
 semaphore.acquire()
 print("run the thread: %s" % n)
 time.sleep(1)
 semaphore.release()

num = 0
semaphore = threading.BoundedSemaphore(5) # 最多允許5個(gè)線程同時(shí)運(yùn)行
for i in range(20):
 t = threading.Thread(target=run, args=(i,))
 t.start()

1.3.4 事件(Event)

類(lèi)名：Event

事件主要提供了三個(gè)方法 set、wait、clear。

事件機(jī)制：全局定義了一個(gè)“Flag”，如果“Flag”的值為False，那么當(dāng)程序執(zhí)行wait方法時(shí)就會(huì)阻塞，如果“Flag”值為T(mén)rue，那么wait方法時(shí)便不再阻塞。這種鎖，類(lèi)似交通紅綠燈（默認(rèn)是紅燈），它屬于在紅燈的時(shí)候一次性阻擋所有線程，在綠燈的時(shí)候，一次性放行所有的排隊(duì)中的線程。
clear：將“Flag”設(shè)置為False
set：將“Flag”設(shè)置為T(mén)rue

import threading

def func(e,i):
 print(i)
 e.wait() # 檢測(cè)當(dāng)前event是什么狀態(tài)，如果是紅燈，則阻塞，如果是綠燈則繼續(xù)往下執(zhí)行。默認(rèn)是紅燈。
 print(i+100)

event = threading.Event()
for i in range(10):
 t = threading.Thread(target=func, args=(event, i))
 t.start()

event.clear() # 主動(dòng)將狀態(tài)設(shè)置為紅燈
inp = input(">>>")
if inp == "1":
 event.set() # 主動(dòng)將狀態(tài)設(shè)置為綠燈

1.3.5 條件(condition)

類(lèi)名：Condition

該機(jī)制會(huì)使得線程等待，只有滿足某條件時(shí)，才釋放n個(gè)線程。

import threading

def condition():
 ret = False
 r = input(">>>")
 if r == "yes":
  ret = True
 return ret

def func(conn, i):
 print(i)
 conn.acquire()
 conn.wait_for(condition) # 這個(gè)方法接受一個(gè)函數(shù)的返回值
 print(i+100)
 conn.release()

c = threading.Condition()
for i in range(10):
 t = threading.Thread(target=func, args=(c, i,))
 t.start()

上面的例子，每輸入一次“yes”放行了一個(gè)線程。下面這個(gè)，可以選擇一次放行幾個(gè)線程。

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import threading

def run(n):
 con.acquire()
 con.wait()
 print("run the thread: %s" %n)
 con.release()

if __name__ == '__main__':

 con = threading.Condition()
 for i in range(10):
  t = threading.Thread(target=run, args=(i,))
  t.start()

 while True:
  inp = input('>>>')
  if inp == "q":
   break
  # 下面這三行是固定語(yǔ)法
  con.acquire()
  con.notify(int(inp)) # 這個(gè)方法接收一個(gè)整數(shù)，表示讓多少個(gè)線程通過(guò)
  con.release()

1.3 全局解釋器鎖（GIL）

既然介紹了多線程和線程鎖，那就不得不提及python的GIL，也就是全局解釋器鎖。在編程語(yǔ)言的世界，python因?yàn)镚IL的問(wèn)題廣受詬病，因?yàn)樗诮忉屍鞯膶用嫦拗屏顺绦蛟谕粫r(shí)間只有一個(gè)線程被CPU實(shí)際執(zhí)行，而不管你的程序里實(shí)際開(kāi)了多少條線程。所以我們經(jīng)常能發(fā)現(xiàn)，python中的多線程編程有時(shí)候效率還不如單線程，就是因?yàn)檫@個(gè)原因。那么，對(duì)于這個(gè)GIL，一些普遍的問(wèn)題如下：

每種編程語(yǔ)言都有GIL嗎？

以python官方Cpython解釋器為代表....其他語(yǔ)言好像未見(jiàn)。

為什么要有GIL？

作為解釋型語(yǔ)言，Python的解釋器必須做到既安全又高效。我們都知道多線程編程會(huì)遇到的問(wèn)題。解釋器要留意的是避免在不同的線程操作內(nèi)部共享的數(shù)據(jù)。同時(shí)它還要保證在管理用戶線程時(shí)總是有最大化的計(jì)算資源。那么，不同線程同時(shí)訪問(wèn)時(shí)，數(shù)據(jù)的保護(hù)機(jī)制是怎樣的呢？答案是解釋器全局鎖GIL。GIL對(duì)諸如當(dāng)前線程狀態(tài)和為垃圾回收而用的堆分配對(duì)象這樣的東西的訪問(wèn)提供著保護(hù)。

為什么不能去掉GIL？

首先，在早期的python解釋器依賴較多的全局狀態(tài)，傳承下來(lái)，使得想要移除當(dāng)今的GIL變得更加困難。其次，對(duì)于程序員而言，僅僅是想要理解它的實(shí)現(xiàn)就需要對(duì)操作系統(tǒng)設(shè)計(jì)、多線程編程、C語(yǔ)言、解釋器設(shè)計(jì)和CPython解釋器的實(shí)現(xiàn)有著非常徹底的理解。

在1999年，針對(duì)Python1.5，一個(gè)“freethreading”補(bǔ)丁已經(jīng)嘗試移除GIL，用細(xì)粒度的鎖來(lái)代替。然而，GIL的移除給單線程程序的執(zhí)行速度帶來(lái)了一定的負(fù)面影響。當(dāng)用單線程執(zhí)行時(shí)，速度大約降低了40%。雖然使用兩個(gè)線程時(shí)在速度上得到了提高，但這個(gè)提高并沒(méi)有隨著核數(shù)的增加而線性增長(zhǎng)。因此這個(gè)補(bǔ)丁沒(méi)有被采納。

另外，在python的不同解釋器實(shí)現(xiàn)中，如PyPy就移除了GIL，其執(zhí)行速度更快（不單單是去除GIL的原因）。然而，我們通常使用的CPython占有著統(tǒng)治地位的使用量，所以，你懂的。

在Python 3.2中實(shí)現(xiàn)了一個(gè)新的GIL，并且?guī)е恍┓e極的結(jié)果。這是自1992年以來(lái)，GIL的一次最主要改變。舊的GIL通過(guò)對(duì)Python指令進(jìn)行計(jì)數(shù)來(lái)確定何時(shí)放棄GIL。在新的GIL實(shí)現(xiàn)中，用一個(gè)固定的超時(shí)時(shí)間來(lái)指示當(dāng)前的線程以放棄這個(gè)鎖。在當(dāng)前線程保持這個(gè)鎖，且當(dāng)?shù)诙€(gè)線程請(qǐng)求這個(gè)鎖的時(shí)候，當(dāng)前線程就會(huì)在5ms后被強(qiáng)制釋放掉這個(gè)鎖（這就是說(shuō)，當(dāng)前線程每5ms就要檢查其是否需要釋放這個(gè)鎖）。當(dāng)任務(wù)是可行的時(shí)候，這會(huì)使得線程間的切換更加可預(yù)測(cè)。

GIL對(duì)我們有什么影響？

最大的影響是我們不能隨意使用多線程。要區(qū)分任務(wù)場(chǎng)景。

在單核cpu情況下對(duì)性能的影響可以忽略不計(jì)，多線程多進(jìn)程都差不多。在多核CPU時(shí)，多線程效率較低。GIL對(duì)單進(jìn)程和多進(jìn)程沒(méi)有影響。

在實(shí)際使用中有什么好的建議？

建議在IO密集型任務(wù)中使用多線程，在計(jì)算密集型任務(wù)中使用多進(jìn)程。深入研究python的協(xié)程機(jī)制，你會(huì)有驚喜的。

更多的詳細(xì)介紹和說(shuō)明請(qǐng)參考下面的文獻(xiàn)：

原文：
譯文：Python 最難的問(wèn)題

1.4 定時(shí)器（Timer）

定時(shí)器，指定n秒后執(zhí)行某操作。很簡(jiǎn)單但很使用的東西。

from threading import Timer
def hello():
 print("hello, world")
t = Timer(1, hello) # 表示1秒后執(zhí)行hello函數(shù)
t.start() 

1.5 隊(duì)列

通常而言，隊(duì)列是一種先進(jìn)先出的數(shù)據(jù)結(jié)構(gòu)，與之對(duì)應(yīng)的是堆棧這種后進(jìn)先出的結(jié)構(gòu)。但是在python中，它內(nèi)置了一個(gè)queue模塊，它不但提供普通的隊(duì)列，還提供一些特殊的隊(duì)列。具體如下：

1. queue.Queue ：先進(jìn)先出隊(duì)列
2. queue.LifoQueue ：后進(jìn)先出隊(duì)列
3. queue.PriorityQueue ：優(yōu)先級(jí)隊(duì)列
4. queue.deque ：雙向隊(duì)列
   

1.5.1 Queue：先進(jìn)先出隊(duì)列

這是最常用也是最普遍的隊(duì)列，先看一個(gè)例子。

import queue
q = queue.Queue(5)
q.put(11)
q.put(22)
q.put(33)

print(q.get())
print(q.get())
print(q.get())

Queue類(lèi)的參數(shù)和方法：

maxsize 隊(duì)列的最大元素個(gè)數(shù)，也就是queue.Queue(5)中的5。當(dāng)隊(duì)列內(nèi)的元素達(dá)到這個(gè)值時(shí)，后來(lái)的元素默認(rèn)會(huì)阻塞，等待隊(duì)列騰出位置。

def __init__(self, maxsize=0):
self.maxsize = maxsize
self._init(maxsize)

qsize() 獲取當(dāng)前隊(duì)列中元素的個(gè)數(shù)，也就是隊(duì)列的大小
empty() 判斷當(dāng)前隊(duì)列是否為空，返回True或者False
full() 判斷當(dāng)前隊(duì)列是否已滿，返回True或者False
put(self, block=True, timeout=None)

往隊(duì)列里放一個(gè)元素，默認(rèn)是阻塞和無(wú)時(shí)間限制的。如果，block設(shè)置為False，則不阻塞，這時(shí)，如果隊(duì)列是滿的，放不進(jìn)去，就會(huì)彈出異常。如果timeout設(shè)置為n秒，則會(huì)等待這個(gè)秒數(shù)后才put，如果put不進(jìn)去則彈出異常。

get(self, block=True, timeout=None)

從隊(duì)列里獲取一個(gè)元素。參數(shù)和put是一樣的意思。

join() 阻塞進(jìn)程，直到所有任務(wù)完成，需要配合另一個(gè)方法task_done。

def join(self):
 with self.all_tasks_done:
  while self.unfinished_tasks:
   self.all_tasks_done.wait()

task_done() 表示某個(gè)任務(wù)完成。每一條get語(yǔ)句后需要一條task_done。

import queue
q = queue.Queue(5)
q.put(11)
q.put(22)
print(q.get())
q.task_done()
print(q.get())
q.task_done()
q.join()

1.5.2 LifoQueue：后進(jìn)先出隊(duì)列

類(lèi)似于“堆?！保筮M(jìn)先出。也較常用。

import queue
q = queue.LifoQueue()
q.put(123)
q.put(456)
print(q.get())

上述代碼運(yùn)行結(jié)果是：456

1.5.3 PriorityQueue：優(yōu)先級(jí)隊(duì)列

帶有權(quán)重的隊(duì)列，每個(gè)元素都是一個(gè)元組，前面的數(shù)字表示它的優(yōu)先級(jí)，數(shù)字越小優(yōu)先級(jí)越高，同樣的優(yōu)先級(jí)先進(jìn)先出

q = queue.PriorityQueue()
q.put((1,"alex1"))
q.put((1,"alex2"))
q.put((1,"alex3"))
q.put((3,"alex3"))
print(q.get())

1.5.4 deque：雙向隊(duì)列

Queue和LifoQueue的“綜合體”，雙向進(jìn)出。方法較多，使用復(fù)雜，慎用！

q = queue.deque()
q.append(123)
q.append(333)
q.appendleft(456)
q.pop()
q.popleft()

1.6 生產(chǎn)者消費(fèi)者模型

利用多線程和隊(duì)列可以搭建一個(gè)生產(chǎn)者消費(fèi)者模型，用于處理大并發(fā)的服務(wù)。

在并發(fā)編程中使用生產(chǎn)者和消費(fèi)者模式能夠解決絕大多數(shù)并發(fā)問(wèn)題。該模式通過(guò)平衡生產(chǎn)線程和消費(fèi)線程的工作能力來(lái)提高程序的整體處理數(shù)據(jù)的速度。

為什么要使用生產(chǎn)者和消費(fèi)者模式

在線程世界里，生產(chǎn)者就是生產(chǎn)數(shù)據(jù)的線程，消費(fèi)者就是消費(fèi)數(shù)據(jù)的線程。在多線程開(kāi)發(fā)當(dāng)中，如果生產(chǎn)者處理速度很快，而消費(fèi)者處理速度很慢，那么生產(chǎn)者就必須等待消費(fèi)者處理完，才能繼續(xù)生產(chǎn)數(shù)據(jù)。同樣的道理，如果消費(fèi)者的處理能力大于生產(chǎn)者，那么消費(fèi)者就必須等待生產(chǎn)者。為了解決這個(gè)問(wèn)題于是引入了生產(chǎn)者和消費(fèi)者模式。

什么是生產(chǎn)者消費(fèi)者模式

生產(chǎn)者消費(fèi)者模式是通過(guò)一個(gè)容器來(lái)解決生產(chǎn)者和消費(fèi)者的強(qiáng)耦合問(wèn)題。生產(chǎn)者和消費(fèi)者彼此之間不直接通訊，而通過(guò)阻塞隊(duì)列來(lái)進(jìn)行通訊，所以生產(chǎn)者生產(chǎn)完數(shù)據(jù)之后不用等待消費(fèi)者處理，直接扔給阻塞隊(duì)列，消費(fèi)者不找生產(chǎn)者要數(shù)據(jù)，而是直接從阻塞隊(duì)列里取，阻塞隊(duì)列就相當(dāng)于一個(gè)緩沖區(qū)，平衡了生產(chǎn)者和消費(fèi)者的處理能力。

這個(gè)阻塞隊(duì)列就是用來(lái)給生產(chǎn)者和消費(fèi)者解耦的?？v觀大多數(shù)設(shè)計(jì)模式，都會(huì)找一個(gè)第三者出來(lái)進(jìn)行解耦，如工廠模式的第三者是工廠類(lèi)，模板模式的第三者是模板類(lèi)。在學(xué)習(xí)一些設(shè)計(jì)模式的過(guò)程中，如果先找到這個(gè)模式的第三者，能幫助我們快速熟悉一個(gè)設(shè)計(jì)模式。

以上摘自方騰飛的《聊聊并發(fā)——生產(chǎn)者消費(fèi)者模式》

下面是一個(gè)簡(jiǎn)單的廚師做包子，顧客吃包子的例子。

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# Author:Liu Jiang
import time
import queue
import threading

q = queue.Queue(10)
def productor(i):
 while True:
  q.put("廚師 %s 做的包子！"%i)
  time.sleep(2)

def consumer(k):
 while True:
  print("顧客 %s 吃了一個(gè) %s"%(k,q.get()))
  time.sleep(1)

for i in range(3):
 t = threading.Thread(target=productor,args=(i,))
 t.start()

for k in range(10):
 v = threading.Thread(target=consumer,args=(k,))
 v.start()

1.7 線程池

在使用多線程處理任務(wù)時(shí)也不是線程越多越好，由于在切換線程的時(shí)候，需要切換上下文環(huán)境，依然會(huì)造成cpu的大量開(kāi)銷(xiāo)。為解決這個(gè)問(wèn)題，線程池的概念被提出來(lái)了。預(yù)先創(chuàng)建好一個(gè)較為優(yōu)化的數(shù)量的線程，讓過(guò)來(lái)的任務(wù)立刻能夠使用，就形成了線程池。在python中，沒(méi)有內(nèi)置的較好的線程池模塊，需要自己實(shí)現(xiàn)或使用第三方模塊。下面是一個(gè)簡(jiǎn)單的線程池：

#!/usr/bin/env python
# -*- coding:utf-8 -*-
# Author:Liu Jiang

import queue
import time
import threading

class MyThreadPool:
 def __init__(self, maxsize=5):
  self.maxsize = maxsize
  self._q = queue.Queue(maxsize)
  for i in range(maxsize):
   self._q.put(threading.Thread)
   
 def get_thread(self):
  return self._q.get()
  
 def add_thread(self):
  self._q.put(threading.Thread)

def task(i, pool):
 print(i)
 time.sleep(1)
 pool.add_thread()

pool = MyThreadPool(5)
for i in range(100):
 t = pool.get_thread()
 obj = t(target=task, args=(i,pool))
 obj.start()

上面的例子是把線程類(lèi)當(dāng)做元素添加到隊(duì)列內(nèi)。實(shí)現(xiàn)方法比較糙，每個(gè)線程使用后就被拋棄，一開(kāi)始就將線程開(kāi)到滿，因此性能較差。下面是一個(gè)相對(duì)好一點(diǎn)的例子：

#!/usr/bin/env python
# -*- coding:utf-8 -*-

import queue
import threading
import contextlib
import time

StopEvent = object() # 創(chuàng)建空對(duì)象

class ThreadPool(object):

 def __init__(self, max_num, max_task_num = None):
  if max_task_num:
   self.q = queue.Queue(max_task_num)
  else:
   self.q = queue.Queue()
  self.max_num = max_num
  self.cancel = False
  self.terminal = False
  self.generate_list = []
  self.free_list = []

 def run(self, func, args, callback=None):
  """
  線程池執(zhí)行一個(gè)任務(wù)
  :param func: 任務(wù)函數(shù)
  :param args: 任務(wù)函數(shù)所需參數(shù)
  :param callback: 任務(wù)執(zhí)行失敗或成功后執(zhí)行的回調(diào)函數(shù)，回調(diào)函數(shù)有兩個(gè)參數(shù)1、任務(wù)函數(shù)執(zhí)行狀態(tài)；2、任務(wù)函數(shù)返回值（默認(rèn)為None，即：不執(zhí)行回調(diào)函數(shù)）
  :return: 如果線程池已經(jīng)終止，則返回True否則None
  """
  if self.cancel:
   return
  if len(self.free_list) == 0 and len(self.generate_list) < self.max_num:
   self.generate_thread()
  w = (func, args, callback,)
  self.q.put(w)

 def generate_thread(self):
  """
  創(chuàng)建一個(gè)線程
  """
  t = threading.Thread(target=self.call)
  t.start()

 def call(self):
  """
  循環(huán)去獲取任務(wù)函數(shù)并執(zhí)行任務(wù)函數(shù)
  """
  current_thread = threading.currentThread
  self.generate_list.append(current_thread)

  event = self.q.get()
  while event != StopEvent:

   func, arguments, callback = event
   try:
    result = func(*arguments)
    success = True
   except Exception as e:
    success = False
    result = None

   if callback is not None:
    try:
     callback(success, result)
    except Exception as e:
     pass

   with self.worker_state(self.free_list, current_thread):
    if self.terminal:
     event = StopEvent
    else:
     event = self.q.get()
  else:

   self.generate_list.remove(current_thread)

 def close(self):
  """
  執(zhí)行完所有的任務(wù)后，所有線程停止
  """
  self.cancel = True
  full_size = len(self.generate_list)
  while full_size:
   self.q.put(StopEvent)
   full_size -= 1

 def terminate(self):
  """
  無(wú)論是否還有任務(wù)，終止線程
  """
  self.terminal = True

  while self.generate_list:
   self.q.put(StopEvent)

  self.q.empty()

 @contextlib.contextmanager
 def worker_state(self, state_list, worker_thread):
  """
  用于記錄線程中正在等待的線程數(shù)
  """
  state_list.append(worker_thread)
  try:
   yield
  finally:
   state_list.remove(worker_thread)

# How to use
pool = ThreadPool(5)

def callback(status, result):
 # status, execute action status
 # result, execute action return value
 pass

def action(i):
 print(i)

for i in range(30):
 ret = pool.run(action, (i,), callback)

time.sleep(5)
print(len(pool.generate_list), len(pool.free_list))
print(len(pool.generate_list), len(pool.free_list))
# pool.close()
# pool.terminate()

二、進(jìn)程

在python中multiprocess模塊提供了Process類(lèi)，實(shí)現(xiàn)進(jìn)程相關(guān)的功能。但是，由于它是基于fork機(jī)制的，因此不被windows平臺(tái)支持。想要在windows中運(yùn)行，必須使用if __name__ == '__main__:的方式，顯然這只能用于調(diào)試和學(xué)習(xí)，不能用于實(shí)際環(huán)境。

（PS：在這里我必須吐槽一下python的包、模塊和類(lèi)的組織結(jié)構(gòu)。在multiprocess中你既可以import大寫(xiě)的Process，也可以import小寫(xiě)的process，這兩者是完全不同的東西。這種情況在python中很多，新手容易傻傻分不清。）

下面是一個(gè)簡(jiǎn)單的多進(jìn)程例子，你會(huì)發(fā)現(xiàn)Process的用法和Thread的用法幾乎一模一樣。

from multiprocessing import Process

def foo(i):
 print("This is Process ", i)

if __name__ == '__main__':
 for i in range(5):
  p = Process(target=foo, args=(i,))
  p.start()

2.1 進(jìn)程的數(shù)據(jù)共享

每個(gè)進(jìn)程都有自己獨(dú)立的數(shù)據(jù)空間，不同進(jìn)程之間通常是不能共享數(shù)據(jù)，創(chuàng)建一個(gè)進(jìn)程需要非常大的開(kāi)銷(xiāo)。

from multiprocessing import Process
list_1 = []
def foo(i):
 list_1.append(i)
 print("This is Process ", i," and list_1 is ", list_1)

if __name__ == '__main__':
 for i in range(5):
  p = Process(target=foo, args=(i,))
  p.start()

 print("The end of list_1:", list_1)

運(yùn)行上面的代碼，你會(huì)發(fā)現(xiàn)列表list_1在各個(gè)進(jìn)程中只有自己的數(shù)據(jù)，完全無(wú)法共享。想要進(jìn)程之間進(jìn)行資源共享可以使用queues/Array/Manager這三個(gè)multiprocess模塊提供的類(lèi)。

2.1.1 使用Array共享數(shù)據(jù)

from multiprocessing import Process
from multiprocessing import Array

def Foo(i,temp):
 temp[0] += 100
 for item in temp:
  print(i,'----->',item)

if __name__ == '__main__':
 temp = Array('i', [11, 22, 33, 44])
 for i in range(2):
  p = Process(target=Foo, args=(i,temp))
  p.start()

對(duì)于Array數(shù)組類(lèi)，括號(hào)內(nèi)的“i”表示它內(nèi)部的元素全部是int類(lèi)型，而不是指字符i，列表內(nèi)的元素可以預(yù)先指定，也可以指定列表長(zhǎng)度。概括的來(lái)說(shuō)就是Array類(lèi)在實(shí)例化的時(shí)候就必須指定數(shù)組的數(shù)據(jù)類(lèi)型和數(shù)組的大小，類(lèi)似temp = Array('i', 5)。對(duì)于數(shù)據(jù)類(lèi)型有下面的表格對(duì)應(yīng)：

'c': ctypes.c_char, 'u': ctypes.c_wchar,
'b': ctypes.c_byte, 'B': ctypes.c_ubyte,
'h': ctypes.c_short, 'H': ctypes.c_ushort,
'i': ctypes.c_int, 'I': ctypes.c_uint,
'l': ctypes.c_long, 'L': ctypes.c_ulong,
'f': ctypes.c_float, 'd': ctypes.c_double

2.1.2 使用Manager共享數(shù)據(jù)

from multiprocessing import Process,Manager

def Foo(i,dic):
 dic[i] = 100+i
 print(dic.values())

if __name__ == '__main__':
 manage = Manager()
 dic = manage.dict()
 for i in range(10):
  p = Process(target=Foo, args=(i,dic))
  p.start()
  p.join()

Manager比Array要好用一點(diǎn)，因?yàn)樗梢酝瑫r(shí)保存多種類(lèi)型的數(shù)據(jù)格式。

2.1.3 使用queues的Queue類(lèi)共享數(shù)據(jù)

import multiprocessing
from multiprocessing import Process
from multiprocessing import queues

def foo(i,arg):
 arg.put(i)
 print('The Process is ', i, "and the queue's size is ", arg.qsize())

if __name__ == "__main__":
 li = queues.Queue(20, ctx=multiprocessing)
 for i in range(10):
  p = Process(target=foo, args=(i,li,))
  p.start()

這里就有點(diǎn)類(lèi)似上面的隊(duì)列了。從運(yùn)行結(jié)果里，你還能發(fā)現(xiàn)數(shù)據(jù)共享中存在的臟數(shù)據(jù)問(wèn)題。另外，比較悲催的是multiprocessing里還有一個(gè)Queue，一樣能實(shí)現(xiàn)這個(gè)功能。

2.2 進(jìn)程鎖

為了防止和多線程一樣的出現(xiàn)數(shù)據(jù)搶奪和臟數(shù)據(jù)的問(wèn)題，同樣需要設(shè)置進(jìn)程鎖。與threading類(lèi)似，在multiprocessing里也有同名的鎖類(lèi)RLock, Lock, Event, Condition, Semaphore，連用法都是一樣樣的?。ㄟ@個(gè)我喜歡）

from multiprocessing import Process
from multiprocessing import queues
from multiprocessing import Array
from multiprocessing import RLock, Lock, Event, Condition, Semaphore
import multiprocessing
import time

def foo(i,lis,lc):
 lc.acquire()
 lis[0] = lis[0] - 1
 time.sleep(1)
 print('say hi',lis[0])
 lc.release()

if __name__ == "__main__":
 # li = []
 li = Array('i', 1)
 li[0] = 10
 lock = RLock()
 for i in range(10):
  p = Process(target=foo,args=(i,li,lock))
  p.start()

2.3 進(jìn)程池

既然有線程池，那必然也有進(jìn)程池。但是，python給我們內(nèi)置了一個(gè)進(jìn)程池，不需要像線程池那樣需要自定義，你只需要簡(jiǎn)單的from multiprocessing import Pool。

#!/usr/bin/env python
# -*- coding:utf-8 -*-
from multiprocessing import Pool
import time

def f1(args):
 time.sleep(1)
 print(args)

if __name__ == '__main__':
 p = Pool(5)
 for i in range(30):
  p.apply_async(func=f1, args= (i,))
 p.close()   # 等子進(jìn)程執(zhí)行完畢后關(guān)閉線程池
 # time.sleep(2)
 # p.terminate()  # 立刻關(guān)閉線程池
 p.join()

進(jìn)程池內(nèi)部維護(hù)一個(gè)進(jìn)程序列，當(dāng)使用時(shí)，去進(jìn)程池中獲取一個(gè)進(jìn)程，如果進(jìn)程池序列中沒(méi)有可供使用的進(jìn)程，那么程序就會(huì)等待，直到進(jìn)程池中有可用進(jìn)程為止。

進(jìn)程池中有以下幾個(gè)主要方法：

1. apply：從進(jìn)程池里取一個(gè)進(jìn)程并執(zhí)行
2. apply_async：apply的異步版本
3. terminate:立刻關(guān)閉線程池
4. join：主進(jìn)程等待所有子進(jìn)程執(zhí)行完畢，必須在close或terminate之后
5. close：等待所有進(jìn)程結(jié)束后，才關(guān)閉線程池

三、協(xié)程

線程和進(jìn)程的操作是由程序觸發(fā)系統(tǒng)接口，最后的執(zhí)行者是系統(tǒng)，它本質(zhì)上是操作系統(tǒng)提供的功能。而協(xié)程的操作則是程序員指定的，在python中通過(guò)yield，人為的實(shí)現(xiàn)并發(fā)處理。

協(xié)程存在的意義：對(duì)于多線程應(yīng)用，CPU通過(guò)切片的方式來(lái)切換線程間的執(zhí)行，線程切換時(shí)需要耗時(shí)。協(xié)程，則只使用一個(gè)線程，分解一個(gè)線程成為多個(gè)“微線程”，在一個(gè)線程中規(guī)定某個(gè)代碼塊的執(zhí)行順序。

協(xié)程的適用場(chǎng)景：當(dāng)程序中存在大量不需要CPU的操作時(shí)（IO）。

在不需要自己“造輪子”的年代，同樣有第三方模塊為我們提供了高效的協(xié)程，這里介紹一下greenlet和gevent。本質(zhì)上，gevent是對(duì)greenlet的高級(jí)封裝，因此一般用它就行，這是一個(gè)相當(dāng)高效的模塊。

在使用它們之前，需要先安裝，可以通過(guò)源碼，也可以通過(guò)pip。

3.1 greenlet

from greenlet import greenlet

def test1():
 print(12)
 gr2.switch()
 print(34)
 gr2.switch()

def test2():
 print(56)
 gr1.switch()
 print(78)

gr1 = greenlet(test1)
gr2 = greenlet(test2)
gr1.switch()

實(shí)際上，greenlet就是通過(guò)switch方法在不同的任務(wù)之間進(jìn)行切換。

3.2 gevent

from gevent import monkey; monkey.patch_all()
import gevent
import requests

def f(url):
 print('GET: %s' % url)
 resp = requests.get(url)
 data = resp.text
 print('%d bytes received from %s.' % (len(data), url))

gevent.joinall([
  gevent.spawn(f, 'https://www.python.org/'),
  gevent.spawn(f, 'https://www.yahoo.com/'),
  gevent.spawn(f, 'https://github.com/'),
])

通過(guò)joinall將任務(wù)f和它的參數(shù)進(jìn)行統(tǒng)一調(diào)度，實(shí)現(xiàn)單線程中的協(xié)程。代碼封裝層次很高，實(shí)際使用只需要了解它的幾個(gè)主要方法即可。

最新評(píng)論