import threading import time def worker(num): """ thread worker function :return: """ time.sleep( 1 ) print ( " Thread %d " % num) return for i in range(20 ): t = threading.Thread(target=worker,args=(i,)，name=“t.%d” % i) t.start()

thread方法說明

t.start() : 激活線程，

t.getName() : 獲取線程的名稱

t.setName() ： 設置線程的名稱

t.name : 獲取或設置線程的名稱

t.is_alive() ： 判斷線程是否為激活狀態

t.isAlive() ：判斷線程是否為激活狀態

t.setDaemon() 設置為後台線程或前台線程（默認：False）;通過一個布爾值設置線程是否為守護線程，必須在執行start()方法之後才可以使用。如果是後台線程，主線程執行過程中，後台線程也在進行，主線程執行完畢後，後台線程不論成功與否，均停止；如果是前台線程，主線程執行過程中，前台線程也在進行，主線程執行完畢後，等待前台線程也執行完成後，程序停止

t.isDaemon() ： 判斷是否為守護線程

t.ident ：獲取線程的標識符。線程標識符是一個非零整數，只有在調用了start()方法之後該屬性才有效，否則它只返回None。

t.join() ：逐個執行每個線程，執行完畢後繼續往下執行，該方法使得多線程變得無意義

t.run() ：線程被cpu調度後自動執行線程對象的run方法

線程鎖threading.RLock和threading.Lock

我們使用線程對數據進行操作的時候，如果多個線程同時修改某個數據，可能會出現不可預料的結果，為了保證數據的準確性，引入了鎖的概念。

例：假設列表A的所有元素就為0，當一個線程從前向後打印列表的所有元素，另外一個線程則從後向前修改列表的元素為1,那麼輸出的時候，列表的元素就會一部分為0，一部分為1,這就導致了數據的不一致。鎖的出現解決了這個問題。

import threading import time globals_num = 0 lock = threading.RLock() def Func(): lock.acquire() # 獲得鎖 global globals_num globals_num += 1 time.sleep( 1 ) print (globals_num) lock.release() # 釋放鎖 for i in range(10 ): t = threading.Thread(target= Func) t.start()

threading.RLock和threading.Lock 的區別

RLock允許在同一線程中被多次acquire。而Lock卻不允許這種情況。如果使用RLock，那麼acquire和release必須成對出現，即調用了n次acquire，必須調用n次的release才能真正釋放所佔用的瑣。

import threading rLock = threading.RLock() # RLock對象rLock.acquire() rLock.acquire() # 在同一線程內，程序不會堵塞。rLock.release() rLock.release()

threading.Event

Event是線程間通信最間的機制之一：一個線程發送一個event信號，其他的線程則等待這個信號。用於主線程控制其他線程的執行。Events 管理一個flag，這個flag可以使用set()設置成True或者使用clear()重置為False，wait()則用於阻塞，在flag為True之前。flag默認為False。

• Event.wait([timeout]) ： 堵塞線程，直到Event對象內部標識位被設為True或超時（如果提供了參數timeout）。

• Event.set() ：將標識位設為Ture

• Event.clear() ： 將標識伴設為False。

• Event.isSet() ：判斷標識位是否為Ture。

import threading def do(event): print ( ' start ' ) event.wait() print ( ' execute ' ) event_obj = threading.Event() for i in range(10 ): t = threading.Thread(target=do, args= (event_obj,)) t.start() event_obj.clear() inp = input( ' input: ' ) if inp == ' true ' : event_obj.set()

當線程執行的時候，如果flag為False，則線程會阻塞，當flag為True的時候，線程不會阻塞。它提供了本地和遠程的並發性。

threading.Condition:

一個condition變量總是與某些類型的鎖相聯繫，這個可以使用默認的情況或創建一個，當幾個condition變量必須共享和同一個鎖的時候，是很有用的。鎖是conditon對象的一部分：沒有必要分別跟踪。

condition變量服從上下文管理協議：with語句塊封閉之前可以獲取與鎖的聯繫。acquire() 和release() 會調用與鎖相關聯的相應的方法。

其他和鎖關聯的方法必須被調用，wait()方法會釋放鎖，當另外一個線程使用notify() or notify_all()喚醒它之前會一直阻塞。一旦被喚醒，wait()會重新獲得鎖並返回，

Condition(lock=None)

Condition類實現了一個conditon變量。這個conditiaon變量允許一個或多個線程等待，直到他們被另一個線程通知。如果lock參數，被給定一個非空的值，，那麼他必須是一個lock或者Rlock對象，它用來做底層鎖。否則，會創建一個新的Rlock對象，用來做底層鎖。

• wait(timeout=None) ： 等待通知，或者等到設定的超時時間。當調用這wait()方法時，如果調用它的線程沒有得到鎖，那麼會拋出一個RuntimeError 異常。wati()釋放鎖以後，在被調用相同條件的另一個進程用notify() or notify_all() 叫醒之前會一直阻塞。wait() 還可以指定一個超時時間。

如果有等待的線程，notify()方法會喚醒一個在等待conditon變量的線程。notify_all() 則會喚醒所有在等待conditon變量的線程。

注意： notify()和notify_all()不會釋放鎖，也就是說，線程被喚醒後不會立刻返回他們的wait() 調用。除非線程調用notify()和notify_all()之後放棄了鎖的所有權。

在典型的設計風格里，利用condition變量用鎖去通許訪問一些共享狀態，線程在獲取到它想得到的狀態前，會反複調用wait()。修改狀態的線程在他們狀態改變時調用notify() or notify_all()，用這種方式，線程會盡可能的獲取到想要的一個等待者狀態。例子： 生產者-消費者模型，

import threading import time def consumer(cond): with cond: print ( " consumer before wait " ) cond.wait() print ( " consumer after wait " ) def producer(cond): with cond: print ( " producer before notifyAll " ) cond.notifyAll() print ( " producer after notifyAll " ) condition = threading.Condition() c1 = threading.Thread(name= " c1 " , target=consumer, args= (condition,)) c2 = threading.Thread(name= " c2 " , target=consumer, args= (condition,)) p = threading.Thread(name= " p " , target=producer, args= (condition,)) c1.start() time.sleep( 2 ) c2.start() time.sleep( 2 ) p.start()

consumer()線程要等待producer()設置了Condition之後才能繼續。

queue模塊

Queue 就是對隊列，它是線程安全的

舉例來說，我們去肯德基吃飯。廚房是給我們做飯的地方，前台負責把廚房做好的飯賣給顧客，顧客則去前台領取做好的飯。這裡的前台就相當於我們的隊列。

這個模型也叫生產者-消費者模型。

import queue q = queue.Queue( maxsize=0 ) #構造一個先進顯出隊列，maxsize指定隊列長度，為0時，表示隊列長度無限制。 q.join() # 等到隊列為kong的時候，在執行別的操作 q.qsize() # 返回隊列的大小（不可靠） q.empty() # 當隊列為空的時候，返回True 否則返回False （不可靠） q.full() # 當隊列滿的時候，返回True，否則返回False （不可靠） q.put( item, block=True, timeout=None ) #將item放入Queue尾部，item必須存在，可以參數block默認為True,表示當隊列滿時，會等待隊列給出可用位置，

　　　　　　　　　　　　　　　　　　　　　　　　 為False時為非阻塞，此時如果隊列已滿，會引發q.get(queue.Full 异常。 可选参数timeout，表示 会阻塞设置的时间，过后，

　　　　　　　　　　　　　　　　　　　　　　　　 如果队列无法给出放入item的位置，则引发 queue.Full 异常

block=True, timeout=None ) #移除並返回隊列頭部的一個值，可選參數block默認為True，表示獲取值的時候，如果隊列為空，則阻塞，為False時，不阻塞，

　　　　　　　　　　　　　　　　　　　　　　若此時隊列為空，則引發queue.Empty異常。可選參數timeout，表示會阻塞設置的時候，過後，如果隊列為空，則引發Empty異常。

q.put_nowait(item) #等效於put(item,block=False)

q.get_nowait() #等效於get(item,block=False)

生產者--消費者：

#!/usr/bin/env python import Queue import threading message = Queue.Queue(10) def producer(i): while True: message.put(i) def consumer(i): while True: msg = message.get() for i in range(12): t = threading.Thread(target=producer, args=(i,)) t.start() for i in range(10): t = threading.Thread(target=consumer, args=(i,)) t.start()

multiprocessing模塊

multiprocessing是python的多進程管理包，和threading.Thread類似。直接從側面用subprocesses替換線程使用GIL的方式，由於這一點，multiprocessing模塊可以讓程序員在給定的機器上充分的利用CPU。

在multiprocessing中，通過創建Process對像生成進程，然後調用它的start()方法，

from multiprocessing import Process def f(name): print ( ' hello ' , name) if __name__ == ' __main__ ' : p = Process(target=f, args=( ' bob ' ,)) p.start() p.join()

進程間的數據共享

在使用並發設計的時候最好盡可能的避免共享數據，尤其是在使用多進程的時候。如果你真有需要要共享數據， multiprocessing提供了兩種方式。

Shared memory

數據可以用Value或Array存儲在一個共享內存地圖裡，如下：

from multiprocessing import Process, Value, Array def f(n, a): n.value = 3.1415927 for i in range(len(a)): a[i] = - a[i] if __name__ == ' __main__ ' : num = Value( ' d ' , 0.0 ) arr = Array( ' i ' , range(10 )) p = Process(target=f, args= (num, arr)) p.start() p.join() print (num.value) print (arr[:])

輸出：

3.1415927 [0, -1, -2, -3, -4, -5, -6, -7, -8, -9]

創建num和arr時，“d”和“i”參數由Array模塊使用的typecodes創建：“d”表示一個雙精度的浮點數，“i”表示一個有符號的整數，這些共享對象將被線程安全的處理。

Array('i', range(10))中的'i'參數：

'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

Server process

由Manager()返回的manager提供list, dict, Namespace, Lock, RLock, Semaphore, BoundedSemaphore, Condition, Event, Barrier, Queue, Value and Array類型的支持。

from multiprocessing import Process, Manager def f(d, l): d[ 1] = ' 1 ' d[ ' 2 ' ] = 2 d[ 0.25] = None l.reverse() if __name__ == ' __main__ ' : with Manager() as manager: d = manager.dict() l = manager.list(range(10 )) p = Process(target=f, args= (d, l)) p.start() p.join() print (d) print (l)

輸出：

{0.25: None, 1: ' 1 ' , ' 2 ' : 2 } [ 9, 8, 7, 6, 5, 4, 3, 2, 1, 0]

Server process manager比shared memory 更靈活，因為它可以支持任意的對像類型。另外，一個單獨的manager可以通過進程在網絡上不同的計算機之間共享，不過他比shared memory要慢。

使用工作池（Using a pool of workers）

Pool類描述了一個工作進程池，他有幾種不同的方法讓任務卸載工作進程。

進程池內部維護一個進程序列，當使用時，則去進程池中獲取一個進程，如果進程池序列中沒有可供使用的進進程，那麼程序就會等待，直到進程池中有可用進程為止。

我們可以用Pool類創建一個進程池， 展開提交的任務給進程池。例：

import asyncio async def cor1(): print ( " COR1 start " ) await cor2() print ( " COR1 end " ) async def cor2(): print ( " COR2 " ) loop = asyncio.get_event_loop() loop.run_until_complete(cor1()) loop.close()

最後三行是重點。

• asyncio.get_event_loop() : asyncio啟動默認的event loop

• run_until_complete() : 這個函數是阻塞執行的，知道所有的異步函數執行完成，

• close() : 關閉event loop。

python 2.7 中的實現：

>>> ret = subprocess.run(["ls", "-l"]) # doesn't capture output CompletedProcess(args=['ls', '-l'], returncode= 0) >>> print (ret.stdout) None >>> subprocess.run("exit 1", shell=True, check= True) Traceback (most recent call last): ... subprocess.CalledProcessError: Command 'exit 1' returned non-zero exit status 1 >>> ret1 = subprocess.run(["ls", "-l", "/dev/null"], stdout= subprocess.PIPE) CompletedProcess(args=['ls', '-l', '/dev/null'], returncode= 0, stdout=b'crw-rw-rw- 1 root root 1, 3 Jan 23 16:23 /dev/null\n' ) >>> print (ret.stdout) b'crw-rw-rw- 1 root root 1, 3 6\xe6\x9c\x88 8 06:50 /dev/null\n'

call方法

subprocess.call ( args, *, stdin = None , stdout = None , stderr = None , shell = False , timeout = None )¶

call()方法等價於：run(..., check=True)和run()方法類所以，只是不支持input參數和check參數;

注意： 不要在這個方法裡使用stdout=PIPE 或stderr=PIPE，當到一個管道的輸出填滿系統的管道緩存時，子進程會被阻塞。

check_call方法

subprocess.check_output(args, *, stdin=None, stderr=None, shell=False, universal_newlines=False, timeout=None)

check_call()方法等價於： run(..., check=True, stdout=PIPE).stdout

3.1新增，3.3時增加了timeout參數，3.4時增加了對關鍵字參數的支持

check_output()方法

內部調用的是run()方法，但是會捕捉到stdout

>>> ret = subprocess.check_output(["ls", "-l", "/dev/null" ]) >>> print (ret) b'crw-rw-rw- 1 root root 1, 3 6\xe6\x9c\x88 8 06:50 /dev/null\n'

Popen類

上面的四個方法本質上調用的都是subprocess中的Popen類。

Popen對像都有以下方法：

poll() ： 檢查子進程是否已經終止，返回一個returncode，相當於exit code。

wait() ： 等待子進程終止，返回一個returncode

communicate(input=None) ：和進程進行交互：發送數據到stdin；從stdout和stderr讀取數據，直到讀取完。等待進程終止。可選參數input會傳遞數據給子進程，當input=None事，沒有數據傳遞給子進程。communicate() 返回一個元組(stdout, stderr). 注意： 讀取的數據在內存的buffer中，所以當數據大小大於buffer或者沒有限制時，不要使用這個方法。