Python多線程、多進程最全整理
作者：錢魏Way來源：https://www.biaodianfu.com/python-multi-thread-and-multi-process.html
在學習Python的過程中，有接觸到多線程編程相關的知識點，先前一直都沒有徹底的搞明白。今天準備花一些時間，把里面的細節(jié)盡可能的梳理清楚。
線程與進程的區(qū)別
進程（process）和線程（thread）是操作系統(tǒng)的基本概念，但是它們比較抽象，不容易掌握。關于多進程和多線程，教科書上最經(jīng)典的一句話是“進程是資源分配的最小單位，線程是CPU調度的最小單位”。線程是程序中一個單一的順序控制流程。進程內一個相對獨立的、可調度的執(zhí)行單元，是系統(tǒng)獨立調度和分派CPU的基本單位指運行中的程序的調度單位。在單個程序中同時運行多個線程完成不同的工作，稱為多線程。
進程和線程區(qū)別
進程是資源分配的基本單位。所有與該進程有關的資源，都被記錄在進程控制塊PCB中。以表示該進程擁有這些資源或正在使用它們。另外，進程也是搶占處理機的調度單位，它擁有一個完整的虛擬地址空間。當進程發(fā)生調度時，不同的進程擁有不同的虛擬地址空間，而同一進程內的不同線程共享同一地址空間。與進程相對應，線程與資源分配無關，它屬于某一個進程，并與進程內的其他線程一起共享進程的資源。線程只由相關堆棧（系統(tǒng)?；蛴脩魲＃┘拇嫫骱途€程控制表TCB組成。寄存器可被用來存儲線程內的局部變量，但不能存儲其他線程的相關變量。通常在一個進程中可以包含若干個線程，它們可以利用進程所擁有的資源。在引入線程的操作系統(tǒng)中，通常都是把進程作為分配資源的基本單位，而把線程作為獨立運行和獨立調度的基本單位。由于線程比進程更小，基本上不擁有系統(tǒng)資源，故對它的調度所付出的開銷就會小得多，能更高效的提高系統(tǒng)內多個程序間并發(fā)執(zhí)行的程度，從而顯著提高系統(tǒng)資源的利用率和吞吐量。因而近年來推出的通用操作系統(tǒng)都引入了線程，以便進一步提高系統(tǒng)的并發(fā)性，并把它視為現(xiàn)代操作系統(tǒng)的一個重要指標。
線程與進程的區(qū)別可以歸納為以下4點：
地址空間和其它資源（如打開文件）：進程間相互獨立，同一進程的各線程間共享。某進程內的線程在其它進程不可見。
通信：進程間通信IPC，線程間可以直接讀寫進程數(shù)據(jù)段（如全局變量）來進行通信——需要進程同步和互斥手段的輔助，以保證數(shù)據(jù)的一致性。
調度和切換：線程上下文切換比進程上下文切換要快得多。
在多線程OS中，進程不是一個可執(zhí)行的實體。
多進程和多線程的比較
| 對比維度 | 多進程 | 多線程 | 總結 |
|---|---|---|---|
| 數(shù)據(jù)共享、同步 | 數(shù)據(jù)共享復雜，同步簡單 | 數(shù)據(jù)共享簡單，同步復雜 | 各有優(yōu)劣 |
| 內存、CPU | 占用內存多，切換復雜，CPU利用率低 | 占用內存少，切換簡單，CPU利用率高 | 線程占優(yōu) |
| 創(chuàng)建、銷毀、切換 | 復雜，速度慢 | 簡單，速度快 | 線程占優(yōu) |
| 編程、調試 | 編程簡單，調試簡單 | 編程復雜，調試復雜 | 進程占優(yōu) |
| 可靠性 | 進程間不會互相影響 | 一個線程掛掉將導致整個進程掛掉 | 進程占優(yōu) |
| 分布式 | 適用于多核、多機，擴展到多臺機器簡單 | 適合于多核 | 進程占優(yōu) |
- 線程在進程下行進（單純的車廂無法運行）
- 一個進程可以包含多個線程（一輛火車可以有多個車廂）
- 不同進程間數(shù)據(jù)很難共享（一輛火車上的乘客很難換到另外一輛火車，比如站點換乘）
- 同一進程下不同線程間數(shù)據(jù)很易共享（A車廂換到B車廂很容易）
- 進程要比線程消耗更多的計算機資源（采用多列火車相比多個車廂更耗資源）
- 進程間不會相互影響，一個線程掛掉將導致整個進程掛掉（一列火車不會影響到另外一列火車，但是如果一列火車上中間的一節(jié)車廂著火了，將影響到該趟火車的所有車廂）
- 進程可以拓展到多機，進程最多適合多核（不同火車可以開在多個軌道上，同一火車的車廂不能在行進的不同的軌道上）
- 進程使用的內存地址可以上鎖，即一個線程使用某些共享內存時，其他線程必須等它結束，才能使用這一塊內存。（比如火車上的洗手間）－”互斥鎖（mutex）”
- 進程使用的內存地址可以限定使用量（比如火車上的餐廳，最多只允許多少人進入，如果滿了需要在門口等，等有人出來了才能進去）－“信號量（semaphore）”
Python全局解釋器鎖GIL
全局解釋器鎖（英語：Global Interpreter Lock，縮寫GIL），并不是Python的特性，它是在實現(xiàn)Python解析器（CPython）時所引入的一個概念。由于CPython是大部分環(huán)境下默認的Python執(zhí)行環(huán)境。所以在很多人的概念里CPython就是Python，也就想當然的把GIL歸結為Python語言的缺陷。那么CPython實現(xiàn)中的GIL又是什么呢？來看看官方的解釋：Python代碼的執(zhí)行由Python 虛擬機(也叫解釋器主循環(huán)，CPython版本)來控制，Python 在設計之初就考慮到要在解釋器的主循環(huán)中，同時只有一個線程在執(zhí)行，即在任意時刻，只有一個線程在解釋器中運行。對Python 虛擬機的訪問由全局解釋器鎖（GIL）來控制，正是這個鎖能保證同一時刻只有一個線程在運行。The mechanism used by the CPython interpreter to assure that only one thread executes Python bytecode at a time. This simplifies the CPython implementation by making the object model (including critical built-in types such as dict) implicitly safe against concurrent access. Locking the entire interpreter makes it easier for the interpreter to be multi-threaded, at the expense of much of the parallelism afforded by multi-processor machines.
GIL 有什么好處？簡單來說，它在單線程的情況更快，并且在和 C 庫結合時更方便，而且不用考慮線程安全問題，這也是早期 Python 最常見的應用場景和優(yōu)勢。另外，GIL的設計簡化了CPython的實現(xiàn)，使得對象模型，包括關鍵的內建類型如字典，都是隱含可以并發(fā)訪問的。鎖住全局解釋器使得比較容易的實現(xiàn)對多線程的支持，但也損失了多處理器主機的并行計算能力。
在多線程環(huán)境中，Python 虛擬機按以下方式執(zhí)行：
- 設置GIL
- 切換到一個線程去運行
- 運行直至指定數(shù)量的字節(jié)碼指令，或者線程主動讓出控制（可以調用sleep(0)）
- 把線程設置為睡眠狀態(tài)
- 解鎖GIL
再次重復以上所有步驟
Python3.2前，GIL的釋放邏輯是當前線程遇見IO操作或者ticks計數(shù)達到100（ticks可以看作是python自身的一個計數(shù)器，專門做用于GIL，每次釋放后歸零，這個計數(shù)可以通過 sys.setcheckinterval 來調整），進行釋放。因為計算密集型線程在釋放GIL之后又會立即去申請GIL，并且通常在其它線程還沒有調度完之前它就已經(jīng)重新獲取到了GIL，就會導致一旦計算密集型線程獲得了GIL，那么它在很長一段時間內都將占據(jù)GIL，甚至一直到該線程執(zhí)行結束。Python 3.2開始使用新的GIL。新的GIL實現(xiàn)中用一個固定的超時時間來指示當前的線程放棄全局鎖。在當前線程保持這個鎖，且其他線程請求這個鎖時，當前線程就會在5毫秒后被強制釋放該鎖。該改進在單核的情況下，對于單個線程長期占用GIL的情況有所好轉。在單核CPU上，數(shù)百次的間隔檢查才會導致一次線程切換。在多核CPU上，存在嚴重的線程顛簸（thrashing）。而每次釋放GIL鎖，線程進行鎖競爭、切換線程，會消耗資源。單核下多線程，每次釋放GIL，喚醒的那個線程都能獲取到GIL鎖，所以能夠無縫執(zhí)行，但多核下，CPU0釋放GIL后，其他CPU上的線程都會進行競爭，但GIL可能會馬上又被CPU0拿到，導致其他幾個CPU上被喚醒后的線程會醒著等待到切換時間后又進入待調度狀態(tài)，這樣會造成線程顛簸(thrashing)，導致效率更低。另外，從上面的實現(xiàn)機制可以推導出，Python的多線程對IO密集型代碼要比CPU密集型代碼更加友好。針對GIL的應對措施：
- 使用更高版本Python（對GIL機制進行了優(yōu)化）
- 使用多進程替換多線程（多進程之間沒有GIL，但是進程本身的資源消耗較多）
- 指定cpu運行線程（使用affinity模塊）
- 使用Jython、IronPython等無GIL解釋器
- 全IO密集型任務時才使用多線程
- 使用協(xié)程（高效的單線程模式，也稱微線程；通常與多進程配合使用）
- 將關鍵組件用C/C++編寫為Python擴展，通過ctypes使Python程序直接調用C語言編譯的動態(tài)鏈接庫的導出函數(shù)。（with nogil調出GIL限制）
Python的多進程包multiprocessing
Python的threading包主要運用多線程的開發(fā)，但由于GIL的存在，Python中的多線程其實并不是真正的多線程，如果想要充分地使用多核CPU的資源，大部分情況需要使用多進程。在Python 2.6版本的時候引入了multiprocessing包，它完整的復制了一套threading所提供的接口方便遷移。唯一的不同就是它使用了多進程而不是多線程。每個進程有自己的獨立的GIL，因此也不會出現(xiàn)進程之間的GIL爭搶。借助這個multiprocessing，你可以輕松完成從單進程到并發(fā)執(zhí)行的轉換。multiprocessing支持子進程、通信和共享數(shù)據(jù)、執(zhí)行不同形式的同步，提供了Process、Queue、Pipe、Lock等組件。Multiprocessing產(chǎn)生的背景
除了應對Python的GIL以外，產(chǎn)生multiprocessing的另外一個原因時Windows操作系統(tǒng)與Linux/Unix系統(tǒng)的不一致。Unix/Linux操作系統(tǒng)提供了一個fork()系統(tǒng)調用，它非常特殊。普通的函數(shù)，調用一次，返回一次，但是fork()調用一次，返回兩次，因為操作系統(tǒng)自動把當前進程（父進程）復制了一份（子進程），然后，分別在父進程和子進程內返回。子進程永遠返回0，而父進程返回子進程的ID。這樣做的理由是，一個父進程可以fork出很多子進程，所以，父進程要記下每個子進程的ID，而子進程只需要調用getpid()就可以拿到父進程的ID。Python的os模塊封裝了常見的系統(tǒng)調用，其中就包括fork，可以在Python程序中輕松創(chuàng)建子進程：import?os
print('Process?(%s)?start...'?%?os.getpid())
\#?Only?works?on?Unix/Linux/Mac:
pid?=?os.fork()
if?pid?==?0:
????print('I?am?child?process?(%s)?and?my?parent?is?%s.'?%?(os.getpid(),?os.getppid()))
else:
????print('I?(%s)?just?created?a?child?process?(%s).'?%?(os.getpid(),?pid))
上述代碼在Linux、Unix和Mac上的執(zhí)行結果為：
Process?(876)?start...
I?(876)?just?created?a?child?process?(877).
I?am?child?process?(877)?and?my?parent?is?876.
multiprocessing常用組件及功能
創(chuàng)建管理進程模塊：
- Process（用于創(chuàng)建進程）
- Pool（用于創(chuàng)建管理進程池）
- Queue（用于進程通信，資源共享）
- Value，Array（用于進程通信，資源共享）
- Pipe（用于管道通信）
- Manager（用于資源共享）
同步子進程模塊：
- Condition（條件變量）
- Event（事件）
- Lock（互斥鎖）
- RLock（可重入的互斥鎖(同一個進程可以多次獲得它，同時不會造成阻塞)
- Semaphore（信號量）
接下來就一起來學習下每個組件及功能的具體使用方法。
Process（用于創(chuàng)建進程）
multiprocessing模塊提供了一個Process類來代表一個進程對象。
在multiprocessing中，每一個進程都用一個Process類來表示。
構造方法：Process([group [, target [, name [, args [, kwargs]]]]])
- group：分組，實際上不使用，值始終為None
- target：表示調用對象，即子進程要執(zhí)行的任務，你可以傳入方法名
- name：為子進程設定名稱
- args：要傳給target函數(shù)的位置參數(shù)，以元組方式進行傳入。
- kwargs：要傳給target函數(shù)的字典參數(shù)，以字典方式進行傳入。
實例方法：
- start()：啟動進程，并調用該子進程中的p.run()
- run()：進程啟動時運行的方法，正是它去調用target指定的函數(shù)，我們自定義類的類中一定要實現(xiàn)該方法
- terminate()：強制終止進程p，不會進行任何清理操作，如果p創(chuàng)建了子進程，該子進程就成了僵尸進程，使用該方法需要特別小心這種情況。如果p還保存了一個鎖那么也將不會被釋放，進而導致死鎖
- is_alive()：返回進程是否在運行。如果p仍然運行，返回True
- join([timeout])：進程同步，主進程等待子進程完成后再執(zhí)行后面的代碼。線程等待p終止（強調：是主線程處于等的狀態(tài)，而p是處于運行的狀態(tài)）。timeout是可選的超時時間（超過這個時間，父線程不再等待子線程，繼續(xù)往下執(zhí)行），需要強調的是，p.join只能join住start開啟的進程，而不能join住run開啟的進程
屬性介紹：
- daemon：默認值為False，如果設為True，代表p為后臺運行的守護進程；當p的父進程終止時，p也隨之終止，并且設定為True后，p不能創(chuàng)建自己的新進程；必須在p.start()之前設置
- name：進程的名稱
- pid：進程的pid
- exitcode：進程在運行時為None、如果為–N，表示被信號N結束(了解即可)
- authkey：進程的身份驗證鍵,默認是由os.urandom()隨機生成的32字符的字符串。這個鍵的用途是為涉及網(wǎng)絡連接的底層進程間通信提供安全性，這類連接只有在具有相同的身份驗證鍵時才能成功（了解即可）
使用示例：（注意：在windows中Process()必須放到if name == ‘main’:下）
from?multiprocessing?import?Process
import?os
def?run_proc(name):
????print('Run?child?process?%s?(%s)...'?%?(name,?os.getpid()))
if?__name__=='__main__':
????print('Parent?process?%s.'?%?os.getpid())
????p?=?Process(target=run_proc,?args=('test',))
????print('Child?process?will?start.')
????p.start()
????p.join()
print('Child?process?end.')
Pool（用于創(chuàng)建管理進程池）
Pool類用于需要執(zhí)行的目標很多，而手動限制進程數(shù)量又太繁瑣時，如果目標少且不用控制進程數(shù)量則可以用Process類。Pool可以提供指定數(shù)量的進程，供用戶調用，當有新的請求提交到Pool中時，如果池還沒有滿，那么就會創(chuàng)建一個新的進程用來執(zhí)行該請求；但如果池中的進程數(shù)已經(jīng)達到規(guī)定最大值，那么該請求就會等待，直到池中有進程結束，就重用進程池中的進程。構造方法：Pool([processes[, initializer[, initargs[, maxtasksperchild[, context]]]]])
- processes ：要創(chuàng)建的進程數(shù)，如果省略，將默認使用cpu_count()返回的數(shù)量。
- initializer：每個工作進程啟動時要執(zhí)行的可調用對象，默認為None。如果initializer是None，那么每一個工作進程在開始的時候會調用initializer(*initargs)。
- initargs：是要傳給initializer的參數(shù)組。
- maxtasksperchild：工作進程退出之前可以完成的任務數(shù)，完成后用一個新的工作進程來替代原進程，來讓閑置的資源被釋放。maxtasksperchild默認是None，意味著只要Pool存在工作進程就會一直存活。
- context: 用在制定工作進程啟動時的上下文，一般使用Pool() 或者一個context對象的Pool()方法來創(chuàng)建一個池，兩種方法都適當?shù)脑O置了context。
- apply(func[, args[, kwargs]])：在一個池工作進程中執(zhí)行func(args,*kwargs),然后返回結果。需要強調的是：此操作并不會在所有池工作進程中并執(zhí)行func函數(shù)。如果要通過不同參數(shù)并發(fā)地執(zhí)行func函數(shù)，必須從不同線程調用p.apply()函數(shù)或者使用p.apply_async()。它是阻塞的。apply很少使用
- apply_async(func[, arg[, kwds={}[, callback=None]]])：在一個池工作進程中執(zhí)行func(args,*kwargs),然后返回結果。此方法的結果是AsyncResult類的實例，callback是可調用對象，接收輸入?yún)?shù)。當func的結果變?yōu)榭捎脮r，將理解傳遞給callback。callback禁止執(zhí)行任何阻塞操作，否則將接收其他異步操作中的結果。它是非阻塞。
- map(func, iterable[, chunksize=None])：Pool類中的map方法，與內置的map函數(shù)用法行為基本一致，它會使進程阻塞直到返回結果。注意，雖然第二個參數(shù)是一個迭代器，但在實際使用中，必須在整個隊列都就緒后，程序才會運行子進程。
- map_async(func, iterable[, chunksize=None])：map_async與map的關系同apply與apply_async
- imap()：imap 與 map的區(qū)別是，map是當所有的進程都已經(jīng)執(zhí)行完了，并將結果返回了，imap()則是立即返回一個iterable可迭代對象。
- imap_unordered()：不保證返回的結果順序與進程添加的順序一致。
- close()：關閉進程池，防止進一步操作。如果所有操作持續(xù)掛起，它們將在工作進程終止前完成。
- join()：等待所有工作進程退出。此方法只能在close()或teminate()之后調用，讓其不再接受新的Process。
- terminate()：結束工作進程，不再處理未處理的任務。
- get()：返回結果，如果有必要則等待結果到達。timeout是可選的。如果在指定時間內還沒有到達，將引發(fā)異常。如果遠程操作中引發(fā)了異常，它將在調用此方法時再次被引發(fā)。
- ready()：如果調用完成，返回True
- successful()：如果調用完成且沒有引發(fā)異常，返回True，如果在結果就緒之前調用此方法，引發(fā)異常
- wait([timeout])：等待結果變?yōu)榭捎谩?/span>
- terminate()：立即終止所有工作進程，同時不執(zhí)行任何清理或結束任何掛起工作。如果p被垃圾回收，將自動調用此函數(shù)
使用示例：
\#?-*-?coding:utf-8?-*-
\#?Pool+map
from?multiprocessing import?Pool
def?test(i):
????print(i)
if?__name__?==?"__main__":
????lists?=?range(100)
????pool?=?Pool(8)
????pool.map(test,?lists)
????pool.close()
????pool.join()
\#?-*-?coding:utf-8?-*-
\#?異步進程池（非阻塞）
from?multiprocessing?import?Pool
def?test(i):
????print(i)
if?__name__?==?"__main__":
????pool?=?Pool(8)
????for?i?in?range(100):
????????'''
??????? For循環(huán)中執(zhí)行步驟：
????????（1）循環(huán)遍歷，將100個子進程添加到進程池（相對父進程會阻塞）
????????（2）每次執(zhí)行8個子進程，等一個子進程執(zhí)行完后，立馬啟動新的子進程。（相對父進程不阻塞）
??????? apply_async為異步進程池寫法。異步指的是啟動子進程的過程，與父進程本身的執(zhí)行（print）是異步的，而For循環(huán)中往進程池添加子進程的過程，與父進程本身的執(zhí)行卻是同步的。
????????'''
????????pool.apply_async(test,?args=(i,))??#?維持執(zhí)行的進程總數(shù)為8，當一個進程執(zhí)行完后啟動一個新進程.
????print("test")
????pool.close()
????pool.join()
\#?-*-?coding:utf-8?-*-
\#?異步進程池（非阻塞）
from?multiprocessing?import?Pool
def?test(i):
????print(i)
if?__name__?==?"__main__":
????pool?=?Pool(8)
????for?i?in?range(100):
????????'''
????????????實際測試發(fā)現(xiàn)，for循環(huán)內部執(zhí)行步驟：
????????????（1）遍歷100個可迭代對象，往進程池放一個子進程
????????????（2）執(zhí)行這個子進程，等子進程執(zhí)行完畢，再往進程池放一個子進程，再執(zhí)行。（同時只執(zhí)行一個子進程）
??????????? for循環(huán)執(zhí)行完畢，再執(zhí)行print函數(shù)。
????????'''
????????pool.apply(test,?args=(i,))??#?維持執(zhí)行的進程總數(shù)為8，當一個進程執(zhí)行完后啟動一個新進程.
????print("test")
????pool.close()
????pool.join()
Queue（用于進程通信，資源共享）
在使用多進程的過程中，最好不要使用共享資源。普通的全局變量是不能被子進程所共享的，只有通過Multiprocessing組件構造的數(shù)據(jù)結構可以被共享。Queue是用來創(chuàng)建進程間資源共享的隊列的類，使用Queue可以達到多進程間數(shù)據(jù)傳遞的功能（缺點：只適用Process類，不能在Pool進程池中使用）。構造方法：Queue([maxsize])- maxsize是隊列中允許最大項數(shù)，省略則無大小限制。
- put()：用以插入數(shù)據(jù)到隊列。put方法還有兩個可選參數(shù)：blocked和timeout。如果blocked為True（默認值），并且timeout為正值，該方法會阻塞timeout指定的時間，直到該隊列有剩余的空間。如果超時，會拋出Queue.Full異常。如果blocked為False，但該Queue已滿，會立即拋出Queue.Full異常。
- get()：可以從隊列讀取并且刪除一個元素。get方法有兩個可選參數(shù)：blocked和timeout。如果blocked為True（默認值），并且timeout為正值，那么在等待時間內沒有取到任何元素，會拋出Queue.Empty異常。如果blocked為False，有兩種情況存在，如果Queue有一個值可用，則立即返回該值，否則，如果隊列為空，則立即拋出Queue.Empty異常。若不希望在empty的時候拋出異常，令blocked為True或者參數(shù)全部置空即可。
- get_nowait()：同q.get(False)
- put_nowait()：同q.put(False)
- empty()：調用此方法時q為空則返回True，該結果不可靠，比如在返回True的過程中，如果隊列中又加入了項目。
- full()：調用此方法時q已滿則返回True，該結果不可靠，比如在返回True的過程中，如果隊列中的項目被取走。
- qsize()：返回隊列中目前項目的正確數(shù)量，結果也不可靠，理由同q.empty()和q.full()一樣
使用示例：
from?multiprocessing?import?Process,?Queue
import?os,?time,?random
def?write(q):
????print('Process?to?write:?%s'?%?os.getpid())
????for?value?in?['A',?'B',?'C']:
????????print('Put?%s?to?queue...'?%?value)
????????q.put(value)
????????time.sleep(random.random())
def?read(q):
????print('Process?to?read:?%s'?%?os.getpid())
????while?True:
????????value?=?q.get(True)
????????print('Get?%s?from?queue.'?%?value)
if?__name__?==?"__main__":
????q?=?Queue()
????pw?=?Process(target=write,?args=(q,))
????pr?=?Process(target=read,?args=(q,))
????pw.start()
????pr.start()
????pw.join()??#?等待pw結束
????pr.terminate()??#?pr進程里是死循環(huán)，無法等待其結束，只能強行終止
- maxsize：隊列中允許最大項數(shù)，省略則無大小限制。
- task_done()：使用者使用此方法發(fā)出信號，表示q.get()的返回項目已經(jīng)被處理。如果調用此方法的次數(shù)大于從隊列中刪除項目的數(shù)量，將引發(fā)ValueError異常
- join():生產(chǎn)者調用此方法進行阻塞，直到隊列中所有的項目均被處理。阻塞將持續(xù)到隊列中的每個項目均調用q.task_done（）方法為止
使用示例：
\#?-*-?coding:utf-8?-*-
from?multiprocessing?import?Process,?JoinableQueue
import?time,?random
def?consumer(q):
????while?True:
????????res?=?q.get()
????????print('消費者拿到了?%s'?%?res)
????????q.task_done()
def?producer(seq,?q):
????for?item?in?seq:
????????time.sleep(random.randrange(1,2))
????????q.put(item)
????????print('生產(chǎn)者做好了?%s'?%?item)
????q.join()
if?__name__?==?"__main__":
????q?=?JoinableQueue()
????seq?=?('產(chǎn)品%s'?%?i?for?i?in?range(5))
????p?=?Process(target=consumer,?args=(q,))
????p.daemon?=?True??#?設置為守護進程，在主線程停止時p也停止，但是不用擔心，producer內調用q.join保證了consumer已經(jīng)處理完隊列中的所有元素
????p.start()
????producer(seq,?q)
????print('主線程')
Value，Array（用于進程通信，資源共享）
multiprocessing 中Value和Array的實現(xiàn)原理都是在共享內存中創(chuàng)建ctypes()對象來達到共享數(shù)據(jù)的目的，兩者實現(xiàn)方法大同小異，只是選用不同的ctypes數(shù)據(jù)類型而已。Value構造方法：Value((typecode_or_type, args[, lock])- typecode_or_type：定義ctypes()對象的類型，可以傳Type code或 C Type，具體對照表見下文。
- args：傳遞給typecode_or_type構造函數(shù)的參數(shù)
- lock：默認為True，創(chuàng)建一個互斥鎖來限制對Value對象的訪問，如果傳入一個鎖，如Lock或RLock的實例，將用于同步。如果傳入False，Value的實例就不會被鎖保護，它將不是進程安全的。
|?Type?code?|?C?Type?????????????|?Python?Type???????|?Minimum?size?in?bytes?|
|?---------?|?------------------?|?-----------------?|?---------------------?|
|?`'b'`?????|?signed?char????????|?int???????????????|?1?????????????????????|
|?`'B'`?????|?unsigned?char??????|?int???????????????|?1?????????????????????|
|?`'u'`?????|?Py_UNICODE?????????|?Unicode?character?|?2?????????????????????|
|?`'h'`?????|?signed?short???????|?int???????????????|?2?????????????????????|
|?`'H'`?????|?unsigned?short?????|?int???????????????|?2?????????????????????|
|?`'i'`?????|?signed?int?????????|?int???????????????|?2?????????????????????|
|?`'I'`?????|?unsigned?int???????|?int???????????????|?2?????????????????????|
|?`'l'`?????|?signed?long????????|?int???????????????|?4?????????????????????|
|?`'L'`?????|?unsigned?long??????|?int???????????????|?4?????????????????????|
|?`'q'`?????|?signed?long?long???|?int???????????????|?8?????????????????????|
|?`'Q'`?????|?unsigned?long?long?|?int???????????????|?8?????????????????????|
|?`'f'`?????|?float??????????????|?float?????????????|?4?????????????????????|
|?`'d'`?????|?double?????????????|?float?????????????|?8?????????????????????|
參考地址：https://docs.python.org/3/library/array.html
Array
構造方法：Array(typecode_or_type, size_or_initializer, **kwds[, lock])
- typecode_or_type：同上
- size_or_initializer：如果它是一個整數(shù)，那么它確定數(shù)組的長度，并且數(shù)組將被初始化為零。否則，size_or_initializer是用于初始化數(shù)組的序列，其長度決定數(shù)組的長度。
- kwds：傳遞給typecode_or_type構造函數(shù)的參數(shù)
- lock：同上
使用示例：
import?multiprocessing
def?f(n,?a):
????n.value?=?3.14
????a[0]?=?5
if?__name__?==?'__main__':
????num?=?multiprocessing.Value('d',?0.0)
????arr?=?multiprocessing.Array('i',?range(10))
????p?=?multiprocessing.Process(target=f,?args=(num,?arr))
????p.start()
????p.join()
????print(num.value)
????print(arr[:])
注意：Value和Array只適用于Process類。
Pipe（用于管道通信）
多進程還有一種數(shù)據(jù)傳遞方式叫管道原理和 Queue相同。Pipe可以在進程之間創(chuàng)建一條管道，并返回元組（conn1,conn2）,其中conn1，conn2表示管道兩端的連接對象，強調一點：必須在產(chǎn)生Process對象之前產(chǎn)生管道。構造方法：Pipe([duplex])- dumplex:默認管道是全雙工的，如果將duplex射成False，conn1只能用于接收，conn2只能用于發(fā)送。
- send(obj)：通過連接發(fā)送對象。obj是與序列化兼容的任意對象
- recv()：接收conn2.send(obj)發(fā)送的對象。如果沒有消息可接收，recv方法會一直阻塞。如果連接的另外一端已經(jīng)關閉，那么recv方法會拋出EOFError。
- close():關閉連接。如果conn1被垃圾回收，將自動調用此方法
- fileno():返回連接使用的整數(shù)文件描述符
- poll([timeout]):如果連接上的數(shù)據(jù)可用，返回True。timeout指定等待的最長時限。如果省略此參數(shù)，方法將立即返回結果。如果將timeout射成None，操作將無限期地等待數(shù)據(jù)到達。
- recv_bytes([maxlength]):接收c.send_bytes()方法發(fā)送的一條完整的字節(jié)消息。maxlength指定要接收的最大字節(jié)數(shù)。如果進入的消息，超過了這個最大值，將引發(fā)IOError異常，并且在連接上無法進行進一步讀取。如果連接的另外一端已經(jīng)關閉，再也不存在任何數(shù)據(jù)，將引發(fā)EOFError異常。
- send_bytes(buffer [, offset [, size]])：通過連接發(fā)送字節(jié)數(shù)據(jù)緩沖區(qū)，buffer是支持緩沖區(qū)接口的任意對象，offset是緩沖區(qū)中的字節(jié)偏移量，而size是要發(fā)送字節(jié)數(shù)。結果數(shù)據(jù)以單條消息的形式發(fā)出，然后調用c.recv_bytes()函數(shù)進行接收
- recv_bytes_into(buffer [, offset]):接收一條完整的字節(jié)消息，并把它保存在buffer對象中，該對象支持可寫入的緩沖區(qū)接口（即bytearray對象或類似的對象）。offset指定緩沖區(qū)中放置消息處的字節(jié)位移。返回值是收到的字節(jié)數(shù)。如果消息長度大于可用的緩沖區(qū)空間，將引發(fā)BufferTooShort異常。
使用示例：
from?multiprocessing?import?Process,?Pipe
import?time
\#?子進程執(zhí)行方法
def?f(Subconn):
????time.sleep(1)
????Subconn.send("吃了嗎")
????print("來自父親的問候:",?Subconn.recv())
????Subconn.close()
if?__name__?==?"__main__":
????parent_conn,?child_conn?=?Pipe()??#?創(chuàng)建管道兩端
????p?=?Process(target=f,?args=(child_conn,))??#?創(chuàng)建子進程
????p.start()
????print("來自兒子的問候:",?parent_conn.recv())
????parent_conn.send("嗯")
Manager（用于資源共享）
Manager()返回的manager對象控制了一個server進程，此進程包含的python對象可以被其他的進程通過proxies來訪問。從而達到多進程間數(shù)據(jù)通信且安全。Manager模塊常與Pool模塊一起使用。Manager支持的類型有l(wèi)ist,dict,Namespace,Lock,RLock,Semaphore,BoundedSemaphore,Condition,Event,Queue,Value和Array。管理器是獨立運行的子進程，其中存在真實的對象，并以服務器的形式運行，其他進程通過使用代理訪問共享對象，這些代理作為客戶端運行。Manager()是BaseManager的子類，返回一個啟動的SyncManager()實例，可用于創(chuàng)建共享對象并返回訪問這些共享對象的代理。BaseManager，創(chuàng)建管理器服務器的基類構造方法：BaseManager([address[, authkey]])- address：(hostname,port),指定服務器的網(wǎng)址地址，默認為簡單分配一個空閑的端口
- authkey：連接到服務器的客戶端的身份驗證，默認為current_process().authkey的值
- start([initializer[, initargs]])：啟動一個單獨的子進程，并在該子進程中啟動管理器服務器
- get_server()：獲取服務器對象
- connect()：連接管理器對象
- shutdown()：關閉管理器對象，只能在調用了start()方法之后調用
- address：只讀屬性，管理器服務器正在使用的地址
- Array(self,*args,**kwds)
- BoundedSemaphore(self,*args,**kwds)
- Condition(self,*args,**kwds)
- Event(self,*args,**kwds)
- JoinableQueue(self,*args,**kwds)
- Lock(self,*args,**kwds)
- Namespace(self,*args,**kwds)
- Pool(self,*args,**kwds)
- Queue(self,*args,**kwds)
- RLock(self,*args,**kwds)
- Semaphore(self,*args,**kwds)
- Value(self,*args,**kwds)
- dict(self,*args,**kwds)
- list(self,*args,**kwds)
使用示例：
import?multiprocessing
def?f(x,?arr,?l,?d,?n):
????x.value?=?3.14
????arr[0]?=?5
????l.append('Hello')
????d[1]?=?2
????n.a?=?10
if?__name__?==?'__main__':
????server?=?multiprocessing.Manager()
????x?=?server.Value('d',?0.0)
????arr?=?server.Array('i',?range(10))
????l?=?server.list()
????d?=?server.dict()
????n?=?server.Namespace()
????proc?=?multiprocessing.Process(target=f,?args=(x,?arr,?l,?d,?n))
????proc.start()
????proc.join()
????print(x.value)
????print(arr)
????print(l)
????print(d)
????print(n)
同步子進程模塊
Lock（互斥鎖）
Lock鎖的作用是當多個進程需要訪問共享資源的時候，避免訪問的沖突。加鎖保證了多個進程修改同一塊數(shù)據(jù)時，同一時間只能有一個修改，即串行的修改，犧牲了速度但保證了數(shù)據(jù)安全。Lock包含兩種狀態(tài)——鎖定和非鎖定，以及兩個基本的方法。構造方法：Lock()
實例方法：
- acquire([timeout]): 使線程進入同步阻塞狀態(tài)，嘗試獲得鎖定。
- release(): 釋放鎖。使用前線程必須已獲得鎖定，否則將拋出異常。
使用示例：
from?multiprocessing?import?Process,?Lock
def?l(lock,?num):
????lock.acquire()
????print("Hello?Num:?%s"?%?(num))
????lock.release()
if?__name__?==?'__main__':
????lock?=?Lock()??#?這個一定要定義為全局
????for?num?in?range(20):
????????Process(target=l,?args=(lock,?num)).start()
RLock（可重入的互斥鎖(同一個進程可以多次獲得它，同時不會造成阻塞)
RLock（可重入鎖）是一個可以被同一個線程請求多次的同步指令。RLock使用了“擁有的線程”和“遞歸等級”的概念，處于鎖定狀態(tài)時，RLock被某個線程擁有。擁有RLock的線程可以再次調用acquire()，釋放鎖時需要調用release()相同次數(shù)?？梢哉J為RLock包含一個鎖定池和一個初始值為0的計數(shù)器，每次成功調用 acquire()/release()，計數(shù)器將+1/-1，為0時鎖處于未鎖定狀態(tài)。
構造方法：RLock()
實例方法：
- acquire([timeout])：同Lock
- release(): 同Lock
Semaphore（信號量）
信號量是一個更高級的鎖機制。信號量內部有一個計數(shù)器而不像鎖對象內部有鎖標識，而且只有當占用信號量的線程數(shù)超過信號量時線程才阻塞。這允許了多個線程可以同時訪問相同的代碼區(qū)。比如廁所有3個坑，那最多只允許3個人上廁所，后面的人只能等里面有人出來了才能再進去，如果指定信號量為3，那么來一個人獲得一把鎖，計數(shù)加1，當計數(shù)等于3時，后面的人均需要等待。一旦釋放，就有人可以獲得一把鎖。構造方法：Semaphore([value])
- value：設定信號量，默認值為1
實例方法：
- acquire([timeout])：同Lock
- release(): 同Lock
使用示例：
from?multiprocessing?import?Process,?Semaphore
import?time,?random
def?go_wc(sem,?user):
????sem.acquire()
????print('%s?占到一個茅坑'?%?user)
????time.sleep(random.randint(0,?3))
????sem.release()
????print(user,?'OK')
if?__name__?==?'__main__':
????sem?=?Semaphore(2)
????p_l?=?[]
????for?i?in?range(5):
????????p?=?Process(target=go_wc,?args=(sem,?'user%s'?%?i,))
????????p.start()
????????p_l.append(p)
????for?i?in?p_l:
????????i.join()
Condition（條件變量）
可以把Condition理解為一把高級的鎖，它提供了比Lock, RLock更高級的功能，允許我們能夠控制復雜的線程同步問題。Condition在內部維護一個鎖對象（默認是RLock），可以在創(chuàng)建Condigtion對象的時候把瑣對象作為參數(shù)傳入。Condition也提供了acquire, release方法，其含義與鎖的acquire, release方法一致，其實它只是簡單的調用內部鎖對象的對應的方法而已。Condition還提供了其他的一些方法。構造方法：Condition([lock/rlock])
- 可以傳遞一個Lock/RLock實例給構造方法，否則它將自己生成一個RLock實例。
- acquire([timeout])：首先進行acquire，然后判斷一些條件。如果條件不滿足則wait
- release()：釋放 Lock
- wait([timeout]): 調用這個方法將使線程進入Condition的等待池等待通知，并釋放鎖。使用前線程必須已獲得鎖定，否則將拋出異常。處于wait狀態(tài)的線程接到通知后會重新判斷條件。
- notify(): 調用這個方法將從等待池挑選一個線程并通知，收到通知的線程將自動調用acquire()嘗試獲得鎖定（進入鎖定池）；其他線程仍然在等待池中。調用這個方法不會釋放鎖定。使用前線程必須已獲得鎖定，否則將拋出異常。
- notifyAll(): 調用這個方法將通知等待池中所有的線程，這些線程都將進入鎖定池嘗試獲得鎖定。調用這個方法不會釋放鎖定。使用前線程必須已獲得鎖定，否則將拋出異常。
使用示例：
import?multiprocessing
import?time
def?stage_1(cond):
????"""perform?first?stage?of?work,
????then?notify?stage_2?to?continue
????"""
????name?=?multiprocessing.current_process().name
????print('Starting',?name)
????with?cond:
????????print('{}?done?and?ready?for?stage?2'.format(name))
????????cond.notify_all()
def?stage_2(cond):
????"""wait?for?the?condition?telling?us?stage_1?is?done"""
????name?=?multiprocessing.current_process().name
????print('Starting',?name)
????with?cond:
????????cond.wait()
????????print('{}?running'.format(name))
if?__name__?==?'__main__':
????condition?=?multiprocessing.Condition()
????s1?=?multiprocessing.Process(name='s1',
?????????????????????????????????target=stage_1,
?????????????????????????????????args=(condition,))
????s2_clients?=?[
????????multiprocessing.Process(
????????????name='stage_2[{}]'.format(i),
????????????target=stage_2,
????????????args=(condition,),
????????)
????????for?i?in?range(1,?3)
????]
????for?c?in?s2_clients:
????????c.start()
????????time.sleep(1)
????s1.start()
????s1.join()
????for?c?in?s2_clients:
????????c.join()
Event（事件）
Event內部包含了一個標志位，初始的時候為false?？梢允褂胹et()來將其設置為true；或者使用clear()將其從新設置為false；可以使用is_set()來檢查標志位的狀態(tài)；另一個最重要的函數(shù)就是wait(timeout=None)，用來阻塞當前線程，直到event的內部標志位被設置為true或者timeout超時。如果內部標志位為true則wait()函數(shù)理解返回。使用示例：
import?multiprocessing
import?time
def?wait_for_event(e):
????"""Wait?for?the?event?to?be?set?before?doing?anything"""
????print('wait_for_event:?starting')
????e.wait()
????print('wait_for_event:?e.is_set()->',?e.is_set())
def?wait_for_event_timeout(e,?t):
????"""Wait?t?seconds?and?then?timeout"""
????print('wait_for_event_timeout:?starting')
????e.wait(t)
????print('wait_for_event_timeout:?e.is_set()->',?e.is_set())
if?__name__?==?'__main__':
????e?=?multiprocessing.Event()
????w1?=?multiprocessing.Process(
????????name='block',
????????target=wait_for_event,
????????args=(e,),
????)
????w1.start()
????w2?=?multiprocessing.Process(
????????name='nonblock',
????????target=wait_for_event_timeout,
????????args=(e,?2),
????)
????w2.start()
????print('main:?waiting?before?calling?Event.set()')
????time.sleep(3)
????e.set()
????print('main:?event?is?set')
其他內容
multiprocessing.dummy 模塊與 multiprocessing 模塊的區(qū)別：dummy 模塊是多線程，而 multiprocessing 是多進程， api 都是通用的。所有可以很方便將代碼在多線程和多進程之間切換。multiprocessing.dummy通常在IO場景可以嘗試使用，比如使用如下方式引入線程池。from?multiprocessing.dummy?import?Pool?as?ThreadPool
multiprocessing.dummy與早期的threading，不同的點好像是在多多核CPU下，只綁定了一個核心（具體未考證）。
參考文檔：
- https://docs.python.org/3/library/multiprocessing.html
- https://www.rddoc.com/doc/Python/3.6.0/zh/library/multiprocessing/
Python并發(fā)之concurrent.futures
Python標準庫為我們提供了threading和multiprocessing模塊編寫相應的多線程/多進程代碼。從Python3.2開始，標準庫為我們提供了concurrent.futures模塊，它提供了ThreadPoolExecutor和ProcessPoolExecutor兩個類，實現(xiàn)了對threading和multiprocessing的更高級的抽象，對編寫線程池/進程池提供了直接的支持。concurrent.futures基礎模塊是executor和future。Executor
Executor是一個抽象類，它不能被直接使用。它為具體的異步執(zhí)行定義了一些基本的方法。ThreadPoolExecutor和ProcessPoolExecutor繼承了Executor，分別被用來創(chuàng)建線程池和進程池的代碼。ThreadPoolExecutor對象
ThreadPoolExecutor類是Executor子類，使用線程池執(zhí)行異步調用。
class?concurrent.futures.ThreadPoolExecutor(max_workers)
使用max_workers數(shù)目的線程池執(zhí)行異步調用。
ProcessPoolExecutor對象
ThreadPoolExecutor類是Executor子類，使用進程池執(zhí)行異步調用。
class?concurrent.futures.ProcessPoolExecutor(max_workers=None)
submit()方法
Executor中定義了submit()方法，這個方法的作用是提交一個可執(zhí)行的回調task，并返回一個future實例。future對象代表的就是給定的調用。
Executor.submit(fn, *args, **kwargs)
- fn：需要異步執(zhí)行的函數(shù)
- *args, **kwargs：fn參數(shù)
使用示例：
from?concurrent?import?futures
def?test(num):
????import?time
????return?time.ctime(),?num
with?futures.ThreadPoolExecutor(max_workers=1)?as?executor:
????future?=?executor.submit(test,?1)
????print(future.result())
map()方法
除了submit，Exectuor還為我們提供了map方法，這個方法返回一個map(func, *iterables)迭代器，迭代器中的回調執(zhí)行返回的結果有序的。Executor.map(func, *iterables, timeout=None)
- func：需要異步執(zhí)行的函數(shù)
- *iterables：可迭代對象，如列表等。每一次func執(zhí)行，都會從iterables中取參數(shù)。
- timeout：設置每次異步操作的超時時間，timeout的值可以是int或float，如果操作超時，會返回raisesTimeoutError；如果不指定timeout參數(shù)，則不設置超時間。
使用示例：
from?concurrent?import?futures
def?test(num):
????import?time
????return?time.ctime(),?num
data?=?[1,?2,?3]
with?futures.ThreadPoolExecutor(max_workers=1)?as?executor:
????for?future?in?executor.map(test,?data):
????????print(future)
shutdown()方法
釋放系統(tǒng)資源,在Executor.submit()或 Executor.map()等異步操作后調用。使用with語句可以避免顯式調用此方法。
Executor.shutdown(wait=True)
Future
Future可以理解為一個在未來完成的操作，這是異步編程的基礎。通常情況下，我們執(zhí)行io操作，訪問url時（如下）在等待結果返回之前會產(chǎn)生阻塞，cpu不能做其他事情，而Future的引入幫助我們在等待的這段時間可以完成其他的操作。Future類封裝了可調用的異步執(zhí)行。Future 實例通過 Executor.submit()方法創(chuàng)建。- cancel()：試圖取消調用。如果調用當前正在執(zhí)行，并且不能被取消，那么該方法將返回False，否則調用將被取消，方法將返回True。
- cancelled()：如果成功取消調用，返回True。
- running()：如果調用當前正在執(zhí)行并且不能被取消，返回True。
- done()：如果調用成功地取消或結束了，返回True。
- result(timeout=None)：返回調用返回的值。如果調用還沒有完成，那么這個方法將等待超時秒。如果調用在超時秒內沒有完成，那么就會有一個Futures.TimeoutError將報出。timeout可以是一個整形或者浮點型數(shù)值，如果timeout不指定或者為None,等待時間無限。如果futures在完成之前被取消了，那么 CancelledError 將會報出。
- exception(timeout=None)：返回調用拋出的異常，如果調用還未完成，該方法會等待timeout指定的時長，如果該時長后調用還未完成，就會報出超時錯誤futures.TimeoutError。timeout可以是一個整形或者浮點型數(shù)值，如果timeout不指定或者為None,等待時間無限。如果futures在完成之前被取消了，那么 CancelledError 將會報出。如果調用完成并且無異常報出，返回None.
- add_done_callback(fn)：將可調用fn捆綁到future上，當Future被取消或者結束運行，fn作為future的唯一參數(shù)將會被調用。如果future已經(jīng)運行完成或者取消，fn將會被立即調用。
- wait(fs, timeout=None, return_when=ALL_COMPLETED)
- 等待fs提供的 Future 實例(possibly created by different Executor instances) 運行結束。返回一個命名的2元集合，分表代表已完成的和未完成的
- return_when 表明什么時候函數(shù)應該返回。它的值必須是一下值之一：
- FIRST_COMPLETED :函數(shù)在任何future結束或者取消的時候返回。
- FIRST_EXCEPTION ：函數(shù)在任何future因為異常結束的時候返回，如果沒有future報錯，效果等于
- ALL_COMPLETED :函數(shù)在所有future結束后才會返回。
- as_completed(fs, timeout=None)：參數(shù)是一個 Future 實例列表，返回值是一個迭代器，在運行結束后產(chǎn)出 Future實例 。
使用示例：
from?concurrent.futures?import?ThreadPoolExecutor,?wait,?as_completed
from?time?import?sleep
from?random?import?randint
def?return_after_5_secs(num):
????sleep(randint(1,?5))
????return?"Return?of?{}".format(num)
pool?=?ThreadPoolExecutor(5)
futures?=?[]
for?x?in?range(5):
????futures.append(pool.submit(return_after_5_secs,?x))
print(1)
for?x?in?as_completed(futures):
????print(x.result())
print(2)
參考鏈接：
- https://pythonhosted.org/futures
猜您喜歡：
?戳我，查看GAN的系列專輯~！附下載 |《TensorFlow 2.0 深度學習算法實戰(zhàn)》
《基于深度神經(jīng)網(wǎng)絡的少樣本學習綜述》
評論
圖片
表情