https://www.cnblogs.com/tkqasn/p/5700281.html
threading — 基于線程的并行

threading用于提供線程相關的操作，線程是應用程序中工作的最小單元。python當前版本的多線程庫沒有實現優先級、線程組，線程也不能被停止、暫停、恢復、中斷。

threading模塊提供的類：

• Thread, Lock, Rlock, Condition, [Bounded]Semaphore, Event, Timer, local。

threading 模塊提供的常用方法：

• threading.current_thread()：獲取當前的線程對象。如果調用方的控制線程不是通過線程模塊創建的，則返回功能有限的虛擬線程對象。
• threading.currentThread(): 返回當前的線程變量。
• threading.get_ident()：獲取線程標識符。標識符還可以通過threading.currentThread().ident獲取
• threading.enumerate(): 返回一個包含正在運行的線程的list。正在運行指線程啟動后、結束前，不包括啟動前和終止后的線程。
• threading.activeCount(): 返回正在運行的線程數量，與len(threading.enumerate())有相同的結果。
• threading.main_thread()：獲取主線程對象

threading 模塊提供的常量：

• threading.TIMEOUT_MAX 設置threading全局超時時間。

Thread類

Thread是線程類，有兩種使用方法，直接傳入要運行的方法或從Thread繼承并覆蓋run()：

# coding:utf-8 import threading import time- 方法一：將要執行的方法作為參數傳給Thread的構造方法 def action(arg):time.sleep(1)print 'the arg is:%s\r' %argfor i in xrange(4):t =threading.Thread(target=action,args=(i,))t.start()print 'main thread end!'- 方法二：從Thread繼承，并重寫run() class MyThread(threading.Thread):def __init__(self,arg):super(MyThread, self).__init__()#注意：一定要顯式的調用父類的初始化函數。self.arg=argdef run(self):#定義每個線程要運行的函數time.sleep(1)print 'the arg is:%s\r' % self.argfor i in xrange(4):t =MyThread(i)t.start()print 'main thread end!'

構造方法：

Thread(group=None, target=None, name=None, args=(), kwargs={})

• group: 線程組，目前還沒有實現，庫引用中提示必須是None；
• target: 要執行的方法；
• name: 線程名；
• args/kwargs: 要傳入方法的參數。

實例方法：

• isAlive(): 返回線程是否在運行。正在運行指啟動后、終止前。

• get/setName(name): 獲取/設置線程名。

• run()：執行start()方法會調用run()，該方將創建Thread對象時傳遞給target的函數名，和傳遞給args、kwargs的參數組合成一個完整的函數，并執行該函數。run()方法一般在自定義Thead類時會用到。
• is/setDaemon(bool): 獲取/設置是后臺線程（默認前臺線程（False））。（在start之前設置）
  如果是后臺線程，主線程執行過程中，后臺線程也在進行，主線程執行完畢后，后臺線程不論成功與否，主線程和后臺線程均停止
  如果是前臺線程，主線程執行過程中，前臺線程也在進行，主線程執行完畢后，等待前臺線程也執行完成后，程序停止
• start(): 啟動線程。
• join([timeout]): 阻塞當前上下文環境的線程，直到調用此方法的線程終止或到達指定的timeout（可選參數）。
• ident：“線程標識符”，如果線程尚未啟動，則為None。如果線程啟動是一個非零整數。
• daemon：如果thread.daemon=True表示該線程為守護線程，必須在調用Start（）之前設置此項，否則將引發RuntimeError。默認為False
• isDaemon()：判斷一個線程是否是守護線程。

使用例子一(未設置setDeamon)：

# coding:utf-8 import threading import timedef action(arg):time.sleep(1)print 'sub thread start!the thread name is:%s' % threading.currentThread().getName()print 'the arg is:%s' %argtime.sleep(1)for i in xrange(4):t =threading.Thread(target=action,args=(i,))t.start()print 'main_thread end!' main_thread end sub thread start. the thread name is:Thread-1sub thread start. the thread name is:Thread-3the arg is:0the arg is:2sub thread start. the thread name is:Thread-2the arg is:1 sub thread start. the thread name is:Thread-4 the arg is:3Process finished with exit code 0

可以看出，創建的4個“前臺”線程，主線程執行過程中，前臺線程也在進行，主線程執行完畢后，等待前臺線程也執行完成后，程序停止

驗證了serDeamon(False)(默認)前臺線程，主線程執行過程中，前臺線程也在進行，主線程執行完畢后，等待前臺線程也執行完成后，主線程停止。

使用例子二（setDeamon=True）

# coding=utf-8 import threading import timedef action(arg):time.sleep(1)print("sub thread start. the thread name is:%s"%threading.currentThread().getName())print("the arg is:%s"%arg)time.sleep(1)for i in range(4):t = threading.Thread(target=action,args=(i,))t.setDaemon(True)t.start()print('main_thread end') main_thread end!Process finished with exit code 0

可以看出，主線程執行完畢后，后臺線程不管是成功與否，主線程均停止
驗證了setDeamon(True)后臺線程，主線程執行過程中，后臺線程也在進行，主線程執行完畢后，后臺線程不論成功與否，主線程均停止。

使用例子三（設置join）

# coding=utf-8 import threading import timedef action(arg):time.sleep(1)print("sub thread start. the thread name is:%s、r"%threading.currentThread().getName())print("the arg is:%s\r"%arg)time.sleep(1)thread_list = []for i in xrange(4):t = threading.Thread(target=action,args=(i,))t.setDaemon(True)thread_list.append(t)for t in thread_list:t.start()for t in thread_list:t.join()print('main_thread end') sub thread start. the thread name is:Thread-2 the arg is:1 sub thread start. the thread name is:Thread-1 the arg is:0 sub thread start. the thread name is:Thread-3 the arg is:2 sub thread start. the thread name is:Thread-4 the arg is:3 main_thread endProcess finished with exit code 0

設置join之后，主線程等待子線程全部執行完成后或者子線程超時后，主線程才結束
復制代碼
驗證了 join()阻塞當前上下文環境的線程，直到調用此方法的線程終止或到達指定的timeout，即使設置了setDeamon（True）主線程依然要等待子線程結束。

使用例子四（join不妥當的用法，使多線程編程順序執行）

# coding=utf-8 import threading import timedef action(arg):time.sleep(1)print("sub thread start. the thread name is:%s\r"%threading.currentThread().getName())print("the arg is:%s\r"%arg)time.sleep(1)for i in xrange(4):t = threading.Thread(target=action,args=(i,))t.setDaemon(True)t.start()t.join()print('main_thread end') sub thread start!the thread name is:Thread-1 the arg is:0 sub thread start!the thread name is:Thread-2 the arg is:1 sub thread start!the thread name is:Thread-3 the arg is:2 sub thread start!the thread name is:Thread-4 the arg is:3 main_thread end!Process finished with exit code 0

可以看出此時，程序只能順序執行，每個線程都被上一個線程的join阻塞，使得“多線程”失去了多線程意義。
復制代碼

Lock、Rlock類

由于線程之間隨機調度：某線程可能在執行n條后，CPU接著執行其他線程。為了多個線程同時操作一個內存中的資源時不產生混亂，我們使用鎖。

Lock（指令鎖）是可用的最低級的同步指令。Lock處于鎖定狀態時，不被特定的線程擁有。Lock包含兩種狀態——鎖定和非鎖定，以及兩個基本的方法。

可以認為Lock有一個鎖定池，當線程請求鎖定時，將線程至于池中，直到獲得鎖定后出池。池中的線程處于狀態圖中的同步阻塞狀態。

RLock（可重入鎖）是一個可以被同一個線程請求多次的同步指令。RLock使用了“擁有的線程”和“遞歸等級”的概念，處于鎖定狀態時，RLock被某個線程擁有。擁有RLock的線程可以再次調用acquire()，釋放鎖時需要調用release()相同次數。

可以認為RLock包含一個鎖定池和一個初始值為0的計數器，每次成功調用 acquire()/release()，計數器將+1/-1，為0時鎖處于未鎖定狀態。

簡言之：Lock屬于全局，Rlock屬于線程。

構造方法：

Lock()，Rlock（）,推薦使用Rlock()

實例方法：

• acquire([timeout]): 嘗試獲得鎖定。使線程進入同步阻塞狀態。
• release(): 釋放鎖。使用前線程必須已獲得鎖定，否則將拋出異常。

例子一（未使用鎖）：

# coding=utf-8 import threading import timeg_num = 0def show(arg):global g_numtime.sleep(1)g_num += 1print g_numfor i in range(10):t = threading.Thread(target=show, args=(i,))t.start()print "main thread stop" main thread stop 1 235 64 6789

多次運行可能產生混亂。這種場景就是適合使用鎖的場景。
復制代碼

例子二（使用鎖）:

# coding:utf-8 import threading import timegl_num = 0 lock = threading.RLock()# 調用acquire([timeout])時，線程將一直阻塞， # 直到獲得鎖定或者直到timeout秒后（timeout參數可選）。 # 返回是否獲得鎖。 def Func():lock.acquire()global gl_numgl_num += 1time.sleep(1)print gl_numlock.release()for i in range(10):t = threading.Thread(target=Func)t.start() 1 2 3 4 5 6 7 8 9 10Process finished with exit code 0

可以看出，全局變量在在每次被調用時都要獲得鎖，才能操作，因此保證了共享數據的安全性
復制代碼

Lock對比Rlock

#coding:utf-8import threading lock = threading.Lock() #Lock對象 lock.acquire() lock.acquire() #產生了死鎖。 lock.release() lock.release() print lock.acquire()import threading rLock = threading.RLock() #RLock對象 rLock.acquire() rLock.acquire() #在同一線程內，程序不會堵塞。 rLock.release() rLock.release()

Condition類

Condition（條件變量）通常與一個鎖關聯。需要在多個Contidion中共享一個鎖時，可以傳遞一個Lock/RLock實例給構造方法，否則它將自己生成一個RLock實例。

可以認為，除了Lock帶有的鎖定池外，Condition還包含一個等待池，池中的線程處于等待阻塞狀態，直到另一個線程調用notify()/notifyAll()通知；得到通知后線程進入鎖定池等待鎖定。

構造方法：

Condition([lock/rlock])

實例方法：

• acquire([timeout])/release(): 調用關聯的鎖的相應方法。
• wait([timeout]): 調用這個方法將使線程進入Condition的等待池等待通知，并釋放鎖。使用前線程必須已獲得鎖定，否則將拋出異常。 當wait接收notify醒來時，已經通過acquire獲得鎖定。小心重復acquire操作.
• notify(): 調用這個方法將從等待池挑選一個線程并通知，收到通知的線程將自動調用acquire()嘗試獲得鎖定（進入鎖定池）；其他線程仍然在等待池中。調用這個方法不會釋放鎖定。使用前線程必須已獲得鎖定，否則將拋出異常。 notify(n)表示通知n個等待的線程
• notifyAll(): 調用這個方法將通知等待池中所有的線程，這些線程都將進入鎖定池嘗試獲得鎖定。調用這個方法不會釋放鎖定。使用前線程必須已獲得鎖定，否則將拋出異常。

例子一：生產者消費者模型

import threading import time# 商品 product = None # 條件變量 con = threading.Condition()# 生產者方法 def produce():global productif con.acquire():while True:if product is None:print 'produce...'product = 'anything'# 通知消費者，商品已經生產con.notify()# 等待通知con.wait()time.sleep(2)# 消費者方法 def consume():global productif con.acquire():while True:if product is not None:print 'consume...'product = None# 通知生產者，商品已經沒了con.notify()# 等待通知con.wait()time.sleep(2)t1 = threading.Thread(target=produce) t2 = threading.Thread(target=consume) t2.start() t1.start() produce... consume... produce... consume... produce... consume... produce... consume... produce... consume...Process finished with exit code -1

程序不斷循環運行下去。重復生產消費過程。
復制代碼

例子二：生產者消費者模型

# coding=utf-8 import threading import time import randomMAX_NUM = 10class Producer(threading.Thread):'''生產者線程類'''def __init__(self, storage, consumeCon, produceCon):super(Producer, self).__init__()self.storage = storageself.consumeCon = consumeConself.produceCon = produceConself.acq_by_wait = Falsedef run(self):while True:self.do_round()time.sleep(1)def do_round(self):'''acq_by_wait標志是否是wait之后被notify喚醒的若是，notify喚醒自動acquire若不是，需要再次主動acquire:return:'''if not self.acq_by_wait:self.produceCon.acquire()if len(self.storage) < MAX_NUM:d = random.randint(1,99)self.storage.append(d)print('_'.join(('produce:', str(threading.currentThread()), str(d), str(len(self.storage)))))self.consumeCon.notify()self.produceCon.release()self.acq_by_wait = Falseelse:print('storage full, waiting...'+str(threading.currentThread()))self.produceCon.wait()self.acq_by_wait = Trueclass Consumer(threading.Thread):'''消費者線程類'''def __init__(self, storage, consumeCon, produceCon):super(Consumer,self).__init__()self.storage = storageself.consumeCon = consumeConself.produceCon = produceConself.acq_by_wait = Falsedef run(self):while True:'''acq_by_wait標志是否wait周被notify喚醒若是，notify喚醒后自動acquire否則，需要再次互動acquire:return: '''if not self.acq_by_wait:self.consumeCon.acquire()if len(self.storage) > 0:print('_'.join((' Consumer:', str(threading.currentThread()), str(self.storage.pop()),str(len(self.storage)))))self.produceCon.notify()self.consumeCon.release()self.acq_by_wait = Falseelse:print(' Consumer waiting...storage empty'+str(threading.currentThread()))self.consumeCon.wait()self.acq_by_wait = Truedef Test():CONSUMER_NUM = 2PRODUCER_NUM = 10storageList = []storageLock = threading.Lock()consumeCon = threading.Condition(lock=storageLock)produceCon = threading.Condition(lock=storageLock)tAllList = []tConsumerList = []for _ in xrange(CONSUMER_NUM):tConsumer = Consumer(storageList, consumeCon, produceCon)tConsumerList.append(tConsumer)tAllList.append(tConsumer)tProducerList = []for _ in xrange(PRODUCER_NUM):tProducer = Producer(storageList, consumeCon, produceCon)tProducerList.append(tProducer)tAllList.append(tProducer)for t in tAllList:t.start()if __name__ == '__main__':Test() Consumer waiting...storage empty<Consumer(Thread-1, started 12572)>Consumer waiting...storage empty<Consumer(Thread-2, started 11944)> produce:_<Producer(Thread-3, started 12584)>_91_1Consumer:_<Consumer(Thread-1, started 12572)>_91_0 produce:_<Producer(Thread-4, started 11940)>_44_1Consumer:_<Consumer(Thread-1, started 12572)>_44_0 produce:_<Producer(Thread-5, started 11228)>_71_1Consumer:_<Consumer(Thread-2, started 11944)>_71_0Consumer waiting...storage empty<Consumer(Thread-1, started 12572)>

例子三：隨機獲取

import threadingalist = None condition = threading.Condition()def doSet():if condition.acquire():while alist is None:condition.wait()for i in range(len(alist))[::-1]:alist[i] = 1condition.release()def doPrint():if condition.acquire():while alist is None:condition.wait()for i in alist:print i,printcondition.release()def doCreate():global alistif condition.acquire():if alist is None:alist = [0 for i in range(10)]condition.notifyAll()condition.release()tset = threading.Thread(target=doSet,name='tset') tprint = threading.Thread(target=doPrint,name='tprint') tcreate = threading.Thread(target=doCreate,name='tcreate') tset.start() tprint.start() tcreate.start() 結果有兩種可能 0 0 0 0 0 0 0 0 0 0 或 1 1 1 1 1 1 1 1 1 1

[Bounded]Semaphore

一個信號量管理一個內部計數器，該計數器因 acquire() 方法的調用而遞減，因 release() 方法的調用而遞增。 計數器的值永遠不會小于零；當 acquire() 方法發現計數器為零時，將會阻塞，直到其它線程調用 release() 方法。

信號量對象也支持 上下文管理協議

構造方法：

threading.Semaphore(value=1)
該類實現信號量對象。信號量對象管理一個原子性的計數器.
可選參數 value 賦予內部計數器初始值，默認值為 1 。如果 value 被賦予小于0的值，將會引發 ValueError 異常。

class threading.BoundedSemaphore(value=1)
該類實現有界信號量。有界信號量通過檢查以確保它當前的值不會超過初始值。如果超過了初始值，將會引發 ValueError 異常。在大多情況下，信號量用于保護數量有限的資源。如果信號量被釋放的次數過多，則表明出現了錯誤。沒有指定時， value 的值默認為1。

實例方法

acquire(blocking=True, timeout=None)
獲取一個信號量。

• 如果內部計數器大于0，將其減1并立即返回True。
• 如果內部計數器為0，則將其阻塞知道被release()調用喚醒。每次release()只喚醒一個線程。

release()
釋放一個信號量，將內部計數器的值增加1。當計數器原先的值為0且有其它線程正在等待它再次大于0時，喚醒正在等待的線程。

示例

信號量主要用在保護有限的資源。以數據庫連接數為例說明，假設當前數據庫支持最大連接數為3，將信號量初始值設為3，那么同時最大可以有三個線程連接數據庫，其他線程若再想連接數據庫，則只有等待，直到某一個線程釋放數據庫連接。

import threading import timesm=threading.Semaphore(3)def connectDb():sm.acquire()print threading.currentThread().getName()+' connecting to db...\n'time.sleep(2)print threading.currentThread().getName()+' released db...\n'sm.release()if __name__ == '__main__':s1=threading.Thread(target=connectDb,args=())s2 = threading.Thread(target=connectDb, args=())s3 = threading.Thread(target=connectDb, args=())s4 = threading.Thread(target=connectDb, args=())s1.start()s2.start()s3.start()s4.start()

結果：

C:\Python27\python.exe E:/pythonproj/基礎練習/t8.py Thread-1 connecting to db...Thread-2 connecting to db...Thread-3 connecting to db...Thread-1 released db...Thread-2 released db... Thread-3 released db...Thread-4 connecting to db...Thread-4 released db...Process finished with exit code 0

Event類

Event（事件）是最簡單的線程通信機制之一：一個線程通知事件，其他線程等待事件。Event內置了一個初始為False的標志，當調用set()時設為True，調用clear()時重置為 False。wait()將阻塞線程至等待阻塞狀態。

Event其實就是一個簡化版的 Condition。Event沒有鎖，無法使線程進入同步阻塞狀態。

構造方法：

Event()

實例方法：

• isSet(): 當內置標志為True時返回True。
• set(): 將標志設為True，并通知所有處于等待阻塞狀態的線程恢復運行狀態。
• clear(): 將標志設為False。
• wait([timeout]): 如果標志為True將立即返回，否則阻塞線程至等待阻塞狀態，等待其他線程調用set()。

例子一

# encoding: UTF-8 import threading import timeevent = threading.Event()def func():# 等待事件，進入等待阻塞狀態print '%s wait for event...' % threading.currentThread().getName()event.wait()# 收到事件后進入運行狀態print '%s recv event.' % threading.currentThread().getName()t1 = threading.Thread(target=func) t2 = threading.Thread(target=func) t1.start() t2.start()time.sleep(2)# 發送事件通知 print 'MainThread set event.' event.set() Thread-1 wait for event... Thread-2 wait for event...#2秒后。。。 MainThread set event. Thread-1 recv event.Thread-2 recv event.

timer類

Timer（定時器）是Thread的派生類，用于在指定時間后調用一個方法。

構造方法：

Timer(interval, function, args=[], kwargs={})

• interval: 指定的時間
• function: 要執行的方法
• args/kwargs: 方法的參數

實例方法：

Timer從Thread派生，沒有增加實例方法。

例子一：

# encoding: UTF-8 import threadingdef func():print 'hello timer!'timer = threading.Timer(5, func) timer.start()

線程延遲5秒后執行。

Barrier類

柵欄類提供一個簡單的同步原語，用于應對固定數量的線程需要彼此相互等待的情況。線程調用 wait() 方法后將阻塞，直到所有線程都調用了 wait() 方法。此時所有線程將被同時釋放。

柵欄對象可以被多次使用，但進程的數量不能改變。

這是一個使用簡便的方法實現客戶端進程與服務端進程同步的例子：

b = Barrier(2, timeout=5)def server():start_server()b.wait()while True:connection = accept_connection()process_server_connection(connection)def client():b.wait()while True:connection = make_connection()process_client_connection(connection)

構造方法

class threading.Barrier(parties, action=None, timeout=None)
創建一個需要 parties 個線程的柵欄對象。如果提供了可調用的 action 參數，它會在所有線程被釋放時在其中一個線程中自動調用。 timeout 是默認的超時時間，如果沒有在 wait() 方法中指定超時時間的話。

實例屬性

wait(timeout=None)
沖出柵欄。當柵欄中所有線程都已經調用了這個函數，它們將同時被釋放。如果提供了 timeout 參數，這里的 timeout 參數優先于創建柵欄對象時提供的 timeout 參數。
函數返回值是一個整數，取值范圍在0到 parties – 1，在每個線程中的返回值不相同。可用于從所有線程中選擇唯一的一個線程執行一些特別的工作。例如：

i = barrier.wait() if i == 0:# Only one thread needs to print thisprint("passed the barrier")

如果創建柵欄對象時在構造函數中提供了 action 參數，它將在其中一個線程釋放前被調用。如果此調用引發了異常，柵欄對象將進入損壞態。
如果發生了超時，柵欄對象將進入破損態。
如果柵欄對象進入破損態，或重置柵欄時仍有線程等待釋放，將會引發 BrokenBarrierError 異常。

reset()
重置柵欄為默認的初始態。如果柵欄中仍有線程等待釋放，這些線程將會收到 BrokenBarrierError 異常。
請注意使用此函數時，如果存在狀態未知的其他線程，則可能需要執行外部同步。 如果柵欄已損壞則最好將其廢棄并新建一個。

abort()
使柵欄處于損壞狀態。 這將導致任何現有和未來對 wait() 的調用失敗并引發 BrokenBarrierError。 例如可以在需要中止某個線程時使用此方法，以避免應用程序的死鎖。
更好的方式是：創建柵欄時提供一個合理的超時時間，來自動避免某個線程出錯。

parties
沖出柵欄所需要的線程數量。

n_waiting
當前時刻正在柵欄中阻塞的線程數量。

broken
一個布爾值，值為 True 表明柵欄為破損態。

exception threading.BrokenBarrierError
異常類，是 RuntimeError 異常的子類，在 Barrier 對象重置時仍有線程阻塞時和對象進入破損態時被引發。

local類

local是一個小寫字母開頭的類，用于管理 thread-local（線程局部的）數據。對于同一個local，線程無法訪問其他線程設置的屬性；線程設置的屬性不會被其他線程設置的同名屬性替換。

可以把local看成是一個“線程-屬性字典”的字典，local封裝了從自身使用線程作為 key檢索對應的屬性字典、再使用屬性名作為key檢索屬性值的細節。

# encoding: UTF-8 import threadinglocal = threading.local() local.tname = 'main'def func():local.tname = 'notmain'print local.tnamet1 = threading.Thread(target=func) t1.start() t1.join()print local.tname notmain main

在 with 語句中使用鎖、條件和信號量

這個模塊提供的帶有 acquire() 和 release() 方法的對象，可以被用作 with 語句的上下文管理器。當進入語句塊時 acquire() 方法會被調用，退出語句塊時 release() 會被調用。因此，以下片段:

with some_lock:# do something...

相當于:

some_lock.acquire() try:# do something... finally:some_lock.release()

現在 Lock 、 RLock 、 Condition 、 Semaphore 和 BoundedSemaphore 對象可以用作 with 語句的上下文管理器。

總結

以上是生活随笔為你收集整理的Python的threading多线程的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。