綜述

Python這門解釋性語(yǔ)言也有專門的線程模型，Python虛擬機(jī)使用GIL（Global Interpreter Lock，全局解釋器鎖）來(lái)互斥線程對(duì)共享資源的訪問(wèn)，但暫時(shí)無(wú)法利用多處理器的優(yōu)勢(shì)。

在Python中我們主要是通過(guò)thread和 threading這兩個(gè)模塊來(lái)實(shí)現(xiàn)的，其中Python的threading模塊是對(duì)thread做了一些包裝的，可以更加方便的被使用，所以我們使用 threading模塊實(shí)現(xiàn)多線程編程。這篇文章我們主要來(lái)看看Python對(duì)多線程編程的支持。

在語(yǔ)言層面，Python對(duì)多線程提供了很好的支持，可以方便地支持創(chuàng)建線程、互斥鎖、信號(hào)量、同步等特性。下面就是官網(wǎng)上介紹threading模塊的基本資料及功能：

實(shí)現(xiàn)模塊

• thread：多線程的底層支持模塊，一般不建議使用；
• threading：對(duì)thread進(jìn)行了封裝，將一些線程的操作對(duì)象化

threading模塊

• Thread 線程類，這是我們用的最多的一個(gè)類，你可以指定線程函數(shù)執(zhí)行或者繼承自它都可以實(shí)現(xiàn)子線程功能；
• Timer與Thread類似，但要等待一段時(shí)間后才開(kāi)始運(yùn)行；
• Lock 鎖原語(yǔ)，這個(gè)我們可以對(duì)全局變量互斥時(shí)使用；
• RLock 可重入鎖，使單線程可以再次獲得已經(jīng)獲得的鎖；
• Condition 條件變量，能讓一個(gè)線程停下來(lái)，等待其他線程滿足某個(gè)“條件”；
• Event 通用的條件變量。多個(gè)線程可以等待某個(gè)事件發(fā)生，在事件發(fā)生后，所有的線程都被激活；
• Semaphore為等待鎖的線程提供一個(gè)類似“等候室”的結(jié)構(gòu)；
• BoundedSemaphore 與semaphore類似，但不允許超過(guò)初始值；
• Queue：實(shí)現(xiàn)了多生產(chǎn)者（Producer）、多消費(fèi)者（Consumer）的隊(duì)列，支持鎖原語(yǔ)，能夠在多個(gè)線程之間提供很好的同步支持。

其中Thread類

• 是你主要的線程類，可以創(chuàng)建進(jìn)程實(shí)例。該類提供的函數(shù)包括：
• getName(self) 返回線程的名字
• isAlive(self) 布爾標(biāo)志，表示這個(gè)線程是否還在運(yùn)行中
• isDaemon(self) 返回線程的daemon標(biāo)志
• join(self, timeout=None) 程序掛起，直到線程結(jié)束，如果給出timeout，則最多阻塞timeout秒
• run(self) 定義線程的功能函數(shù)
• setDaemon(self, daemonic) 把線程的daemon標(biāo)志設(shè)為daemonic
• setName(self, name) 設(shè)置線程的名字
• start(self) 開(kāi)始線程執(zhí)行

其中Queue提供的類

• Queue隊(duì)列
• LifoQueue后入先出（LIFO）隊(duì)列
• PriorityQueue 優(yōu)先隊(duì)列

接下來(lái)，我們將會(huì)用一個(gè)一個(gè)示例來(lái)展示threading的各個(gè)功能，包括但不限于：兩種方式起線程、threading.Thread類的重要函數(shù)、使用Lock互斥及RLock實(shí)現(xiàn)重入鎖、使用Condition實(shí)現(xiàn)生產(chǎn)者和消費(fèi)者模型、使用Event和Semaphore多線程通信

兩種方式起線程

在Python中我們主要是通過(guò)thread和threading這兩個(gè)模塊來(lái)實(shí)現(xiàn)的，其中Python的threading模塊是對(duì)thread做了一些包裝的，可以更加方便的被使用，所以我們使用threading模塊實(shí)現(xiàn)多線程編程。一般來(lái)說(shuō)，使用線程有兩種模式，一種是創(chuàng)建線程要執(zhí)行的函數(shù)，把這個(gè)函數(shù)傳遞進(jìn)Thread對(duì)象里，讓它來(lái)執(zhí)行；另一種是直接從Thread繼承，創(chuàng)建一個(gè)新的class，把線程執(zhí)行的代碼放到這個(gè)新的 class里。

將函數(shù)傳遞進(jìn)Thread對(duì)象：

import threadingdef thread_fun(num): for n in range(0, int(num)): print " I come from %s, num: %s" %( threading.currentThread().getName(), n) def main(thread_num): thread_list = list(); # 先創(chuàng)建線程對(duì)象 for i in range(0, thread_num): thread_name = "thread_%s" %i thread_list.append(threading.Thread(target = thread_fun, name = thread_name, args = (20,))) # 啟動(dòng)所有線程 for thread in thread_list: thread.start() # 主線程中等待所有子線程退出 for thread in thread_list: thread.join() if __name__ == "__main__": main(3)

程序啟動(dòng)了3個(gè)線程，并且打印了每一個(gè)線程的線程名字，這個(gè)比較簡(jiǎn)單吧，處理重復(fù)任務(wù)就派出用場(chǎng)了，下面介紹使用繼承threading的方式；

繼承自threading.Thread類:

import threadingclass MyThread(threading.Thread): def __init__(self): threading.Thread.__init__(self); def run(self): print "I am %s" %self.name if __name__ == "__main__": for thread in range(0, 5): t = MyThread() t.start()

接下來(lái)，將會(huì)介紹如何控制這些線程，包括子線程的退出，子線程是否存活及將子線程設(shè)置為守護(hù)線程(Daemon)。

threading.Thread類的重要函數(shù)

介紹threading模塊中的主類Thread的一些主要方法，實(shí)例代碼如下：

import threadingclass MyThread(threading.Thread): def __init__(self): threading.Thread.__init__(self) def run(self): print "I am %s" % (self.name) if __name__ == "__main__": for i in range(0, 5): my_thread = MyThread() my_thread.start()

1、name相關(guān)
你可以為每一個(gè)thread指定name，默認(rèn)的是Thread-No形式的，如上述實(shí)例代碼打印出的一樣：

I am Thread-1 I am Thread-2 I am Thread-3 I am Thread-4 I am Thread-5

當(dāng)然你可以指定每一個(gè)thread的name，這個(gè)通過(guò)setName方法，代碼：

def __init__(self): threading.Thread.__init__(self) self.setName("new" + self.name)

2、join方法
join方法原型如下，這個(gè)方法是用來(lái)阻塞當(dāng)前上下文，直至該線程運(yùn)行結(jié)束：

def join(self, timeout=None):

timeout可以設(shè)置超時(shí)時(shí)間

3、setDaemon方法
當(dāng)我們?cè)诔绦蜻\(yùn)行中，執(zhí)行一個(gè)主線程，如果主線程又創(chuàng)建一個(gè)子線程，主線程和子線程就分兵兩路，當(dāng)主線程完成想退出時(shí)，會(huì)檢驗(yàn)子線程是否完成。如果子線程未完成，則主線程會(huì)等待子線程完成后再退出。但是有時(shí)候我們需要的是，只要主線程完成了，不管子線程是否完成，都要和主線程一起退出，這時(shí)就可以用setDaemon方法，并設(shè)置其參數(shù)為True。

使用Lock互斥鎖

現(xiàn)在我們考慮這樣一個(gè)問(wèn)題：假設(shè)各個(gè)線程需要訪問(wèn)同一公共資源，我們的代碼該怎么寫？

import threading import time counter = 0 class MyThread(threading.Thread): def __init__(self): threading.Thread.__init__(self) def run(self): global counter time.sleep(1); counter += 1 print "I am %s, set counter:%s" % (self.name, counter) if __name__ == "__main__": for i in range(0, 200): my_thread = MyThread() my_thread.start()

解決上面的問(wèn)題，我們興許會(huì)寫出這樣的代碼，我們假設(shè)跑200個(gè)線程，但是這200個(gè)線程都會(huì)去訪問(wèn)counter這個(gè)公共資源，并對(duì)該資源進(jìn)行處理(counter += 1)，代碼看起來(lái)就是這個(gè)樣了，但是我們看下運(yùn)行結(jié)果：

I am Thread-69, set counter:64 I am Thread-73, set counter:66I am Thread-74, set counter:67I am Thread-75, set counter:68I am Thread-76, set counter:69I am Thread-78, set counter:70I am Thread-77, set counter:71I am Thread-58, set counter:72I am Thread-60, set counter:73I am Thread-62, set counter:74I am Thread-66, set counter:75I am Thread-70, set counter:76I am Thread-72, set counter:77I am Thread-79, set counter:78I am Thread-71, set counter:78

打印結(jié)果我只貼了一部分，從中我們已經(jīng)看出了這個(gè)全局資源(counter)被搶占的情況，問(wèn)題產(chǎn)生的原因就是沒(méi)有控制多個(gè)線程對(duì)同一資源的訪問(wèn)，對(duì)數(shù)據(jù)造成破壞，使得線程運(yùn)行的結(jié)果不可預(yù)期。這種現(xiàn)象稱為“線程不安全”。在開(kāi)發(fā)過(guò)程中我們必須要避免這種情況，那怎么避免？這就用到了我們?cè)诰C述中提到的互斥鎖了。

互斥鎖概念

Python編程中，引入了對(duì)象互斥鎖的概念，來(lái)保證共享數(shù)據(jù)操作的完整性。每個(gè)對(duì)象都對(duì)應(yīng)于一個(gè)可稱為” 互斥鎖” 的標(biāo)記，這個(gè)標(biāo)記用來(lái)保證在任一時(shí)刻，只能有一個(gè)線程訪問(wèn)該對(duì)象。在Python中我們使用threading模塊提供的Lock類。
我們對(duì)上面的程序進(jìn)行整改，為此我們需要添加一個(gè)互斥鎖變量mutex = threading.Lock()，然后在爭(zhēng)奪資源的時(shí)候之前我們會(huì)先搶占這把鎖mutex.acquire()，對(duì)資源使用完成之后我們?cè)卺尫胚@把鎖mutex.release()。代碼如下：

import threading import time counter = 0 mutex = threading.Lock() class MyThread(threading.Thread): def __init__(self): threading.Thread.__init__(self) def run(self): global counter, mutex time.sleep(1); if mutex.acquire(): counter += 1 print "I am %s, set counter:%s" % (self.name, counter) mutex.release() if __name__ == "__main__": for i in range(0, 100): my_thread = MyThread() my_thread.start()

同步阻塞
當(dāng)一個(gè)線程調(diào)用Lock對(duì)象的acquire()方法獲得鎖時(shí)，這把鎖就進(jìn)入“l(fā)ocked”狀態(tài)。因?yàn)槊看沃挥幸粋€(gè)線程1可以獲得鎖，所以如果此時(shí)另一個(gè)線程2試圖獲得這個(gè)鎖，該線程2就會(huì)變?yōu)椤癰lock“同步阻塞狀態(tài)。直到擁有鎖的線程1調(diào)用鎖的release()方法釋放鎖之后，該鎖進(jìn)入“unlocked”狀態(tài)。線程調(diào)度程序從處于同步阻塞狀態(tài)的線程中選擇一個(gè)來(lái)獲得鎖，并使得該線程進(jìn)入運(yùn)行（running）狀態(tài)。

進(jìn)一步考慮

通過(guò)對(duì)公共資源使用互斥鎖，這樣就簡(jiǎn)單的到達(dá)了我們的目的，但是如果我們又遇到下面的情況：

• 1、遇到鎖嵌套的情況該怎么辦，這個(gè)嵌套是指當(dāng)我一個(gè)線程在獲取臨界資源時(shí)，又需要再次獲取；
• 2、如果有多個(gè)公共資源，在線程間共享多個(gè)資源的時(shí)候，如果兩個(gè)線程分別占有一部分資源并且同時(shí)等待對(duì)方的資源；

上述這兩種情況會(huì)直接造成程序掛起，即死鎖，下面我們會(huì)談死鎖及可重入鎖RLock。

死鎖的形成

前一篇文章Python：使用threading模塊實(shí)現(xiàn)多線程編程四[使用Lock互斥鎖]我們已經(jīng)開(kāi)始涉及到如何使用互斥鎖來(lái)保護(hù)我們的公共資源了，現(xiàn)在考慮下面的情況–
如果有多個(gè)公共資源，在線程間共享多個(gè)資源的時(shí)候，如果兩個(gè)線程分別占有一部分資源并且同時(shí)等待對(duì)方的資源，這會(huì)引起什么問(wèn)題？

死鎖概念
所謂死鎖： 是指兩個(gè)或兩個(gè)以上的進(jìn)程在執(zhí)行過(guò)程中，因爭(zhēng)奪資源而造成的一種互相等待的現(xiàn)象，若無(wú)外力作用，它們都將無(wú)法推進(jìn)下去。此時(shí)稱系統(tǒng)處于死鎖狀態(tài)或系統(tǒng)產(chǎn)生了死鎖，這些永遠(yuǎn)在互相等待的進(jìn)程稱為死鎖進(jìn)程。 由于資源占用是互斥的，當(dāng)某個(gè)進(jìn)程提出申請(qǐng)資源后，使得有關(guān)進(jìn)程在無(wú)外力協(xié)助下，永遠(yuǎn)分配不到必需的資源而無(wú)法繼續(xù)運(yùn)行，這就產(chǎn)生了一種特殊現(xiàn)象死鎖。

import threadingcounterA = 0 counterB = 0 mutexA = threading.Lock() mutexB = threading.Lock() class MyThread(threading.Thread): def __init__(self): threading.Thread.__init__(self) def run(self): self.fun1() self.fun2() def fun1(self): global mutexA, mutexB if mutexA.acquire(): print "I am %s , get res: %s" %(self.name, "ResA") if mutexB.acquire(): print "I am %s , get res: %s" %(self.name, "ResB") mutexB.release() mutexA.release() def fun2(self): global mutexA, mutexB if mutexB.acquire(): print "I am %s , get res: %s" %(self.name, "ResB") if mutexA.acquire(): print "I am %s , get res: %s" %(self.name, "ResA") mutexA.release() mutexB.release() if __name__ == "__main__": for i in range(0, 100): my_thread = MyThread() my_thread.start()

代碼中展示了一個(gè)線程的兩個(gè)功能函數(shù)分別在獲取了一個(gè)競(jìng)爭(zhēng)資源之后再次獲取另外的競(jìng)爭(zhēng)資源，我們看運(yùn)行結(jié)果：

I am Thread-1 , get res: ResA I am Thread-1 , get res: ResB I am Thread-2 , get res: ResAI am Thread-1 , get res: ResB

可以看到，程序已經(jīng)掛起在那兒了，這種現(xiàn)象我們就稱之為”死鎖“。

避免死鎖
避免死鎖主要方法就是：正確有序的分配資源，避免死鎖算法中最有代表性的算法是Dijkstra E.W 于1968年提出的銀行家算法

可重入鎖RLock

考慮這種情況：如果一個(gè)線程遇到鎖嵌套的情況該怎么辦，這個(gè)嵌套是指當(dāng)我一個(gè)線程在獲取臨界資源時(shí)，又需要再次獲取。
根據(jù)這種情況，代碼如下：

import threading import time counter = 0 mutex = threading.Lock() class MyThread(threading.Thread): def __init__(self): threading.Thread.__init__(self) def run(self): global counter, mutex time.sleep(1); if mutex.acquire(): counter += 1 print "I am %s, set counter:%s" % (self.name, counter) if mutex.acquire(): counter += 1 print "I am %s, set counter:%s" % (self.name, counter) mutex.release() mutex.release() if __name__ == "__main__": for i in range(0, 200): my_thread = MyThread() my_thread.start()

這種情況的代碼運(yùn)行情況如下：

I am Thread-1, set counter:1

之后就直接掛起了，這種情況形成了最簡(jiǎn)單的死鎖。
那有沒(méi)有一種情況可以在某一個(gè)線程使用互斥鎖訪問(wèn)某一個(gè)競(jìng)爭(zhēng)資源時(shí)，可以再次獲取呢？在Python中為了支持在同一線程中多次請(qǐng)求同一資源，python提供了“可重入鎖”：threading.RLock。這個(gè)RLock內(nèi)部維護(hù)著一個(gè)Lock和一個(gè)counter變量，counter記錄了acquire的次數(shù)，從而使得資源可以被多次require。直到一個(gè)線程所有的acquire都被release，其他的線程才能獲得資源。上面的例子如果使用RLock代替Lock，則不會(huì)發(fā)生死鎖：

代碼只需將上述的：

mutex = threading.Lock()

替換成：

mutex = threading.RLock()

使用Condition實(shí)現(xiàn)復(fù)雜同步

目前我們已經(jīng)會(huì)使用Lock去對(duì)公共資源進(jìn)行互斥訪問(wèn)了，也探討了同一線程可以使用RLock去重入鎖，但是盡管如此我們只不過(guò)才處理了一些程序中簡(jiǎn)單的同步現(xiàn)象，我們甚至還不能很合理的去解決使用Lock鎖帶來(lái)的死鎖問(wèn)題。所以我們得學(xué)會(huì)使用更深層的解決同步問(wèn)題。

Python提供的Condition對(duì)象提供了對(duì)復(fù)雜線程同步問(wèn)題的支持。Condition被稱為條件變量，除了提供與Lock類似的acquire和release方法外，還提供了wait和notify方法。

使用Condition的主要方式為：線程首先acquire一個(gè)條件變量，然后判斷一些條件。如果條件不滿足則wait；如果條件滿足，進(jìn)行一些處理改變條件后，通過(guò)notify方法通知其他線程，其他處于wait狀態(tài)的線程接到通知后會(huì)重新判斷條件。不斷的重復(fù)這一過(guò)程，從而解決復(fù)雜的同步問(wèn)題。

下面我們通過(guò)很著名的“生產(chǎn)者-消費(fèi)者”模型來(lái)來(lái)演示下，在Python中使用Condition實(shí)現(xiàn)復(fù)雜同步。

import threading import time condition = threading.Condition() products = 0 class Producer(threading.Thread): def __init__(self): threading.Thread.__init__(self) def run(self): global condition, products while True: if condition.acquire(): if products < 10: products += 1; print "Producer(%s):deliver one, now products:%s" %(self.name, products) condition.notify() else: print "Producer(%s):already 10, stop deliver, now products:%s" %(self.name, products) condition.wait(); condition.release() time.sleep(2) class Consumer(threading.Thread): def __init__(self): threading.Thread.__init__(self) def run(self): global condition, products while True: if condition.acquire(): if products > 1: products -= 1 print "Consumer(%s):consume one, now products:%s" %(self.name, products) condition.notify() else: print "Consumer(%s):only 1, stop consume, products:%s" %(self.name, products) condition.wait(); condition.release() time.sleep(2) if __name__ == "__main__": for p in range(0, 2): p = Producer() p.start() for c in range(0, 10): c = Consumer() c.start()

代碼中主要實(shí)現(xiàn)了生產(chǎn)者和消費(fèi)者線程，雙方將會(huì)圍繞products來(lái)產(chǎn)生同步問(wèn)題，首先是2個(gè)生成者生產(chǎn)products ，而接下來(lái)的10個(gè)消費(fèi)者將會(huì)消耗products，代碼運(yùn)行如下：

Producer(Thread-1):deliver one, now products:1 Producer(Thread-2):deliver one, now products:2 Consumer(Thread-3):consume one, now products:1 Consumer(Thread-4):only 1, stop consume, products:1 Consumer(Thread-5):only 1, stop consume, products:1 Consumer(Thread-6):only 1, stop consume, products:1 Consumer(Thread-7):only 1, stop consume, products:1 Consumer(Thread-8):only 1, stop consume, products:1 Consumer(Thread-10):only 1, stop consume, products:1 Consumer(Thread-9):only 1, stop consume, products:1 Consumer(Thread-12):only 1, stop consume, products:1 Consumer(Thread-11):only 1, stop consume, products:1

另外：Condition對(duì)象的構(gòu)造函數(shù)可以接受一個(gè)Lock/RLock對(duì)象作為參數(shù)，如果沒(méi)有指定，則Condition對(duì)象會(huì)在內(nèi)部自行創(chuàng)建一個(gè)RLock；除了notify方法外，Condition對(duì)象還提供了notifyAll方法，可以通知waiting池中的所有線程嘗試acquire內(nèi)部鎖。由于上述機(jī)制，處于waiting狀態(tài)的線程只能通過(guò)notify方法喚醒，所以notifyAll的作用在于防止有線程永遠(yuǎn)處于沉默狀態(tài)。

使用Event實(shí)現(xiàn)線程間通信

使用threading.Event可以實(shí)現(xiàn)線程間相互通信，之前的Python：使用threading模塊實(shí)現(xiàn)多線程編程七[使用Condition實(shí)現(xiàn)復(fù)雜同步]我們已經(jīng)初步實(shí)現(xiàn)了線程間通信的基本功能，但是更為通用的一種做法是使用threading.Event對(duì)象。
使用threading.Event可以使一個(gè)線程等待其他線程的通知，我們把這個(gè)Event傳遞到線程對(duì)象中，Event默認(rèn)內(nèi)置了一個(gè)標(biāo)志，初始值為False。一旦該線程通過(guò)wait()方法進(jìn)入等待狀態(tài)，直到另一個(gè)線程調(diào)用該Event的set()方法將內(nèi)置標(biāo)志設(shè)置為True時(shí)，該Event會(huì)通知所有等待狀態(tài)的線程恢復(fù)運(yùn)行。

import threading import time class MyThread(threading.Thread): def __init__(self, signal): threading.Thread.__init__(self) self.singal = signal def run(self): print "I am %s,I will sleep ..."%self.name self.singal.wait() print "I am %s, I awake..." %self.name if __name__ == "__main__": singal = threading.Event() for t in range(0, 3): thread = MyThread(singal) thread.start() print "main thread sleep 3 seconds... " time.sleep(3) singal.set()

運(yùn)行效果如下：

I am Thread-1,I will sleep ... I am Thread-2,I will sleep ... I am Thread-3,I will sleep ... main thread sleep 3 seconds... I am Thread-1, I awake...I am Thread-2, I awake... I am Thread-3, I awake...


文章來(lái)源：https://my.oschina.net/u/3041656/blog/794357

轉(zhuǎn)載于:https://www.cnblogs.com/ranxf/p/8205494.html

總結(jié)

以上是生活随笔為你收集整理的python_threading模块实现多线程详解（转）的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。