詳解Python多線程、多進程

在學習Python的過程中，有接觸到多線程編程相關的知識點，先前一直都沒有徹底的搞明白。今天準備花一些時間，把里面的細節(jié)盡可能的梳理清楚。

線程與進程的區(qū)別

進程（process）和線程（thread）是操作系統(tǒng)的基本概念，但是它們比較抽象，不容易掌握。關于多進程和多線程，教科書上最經(jīng)典的一句話是“進程是資源分配的最小單位，線程是CPU調(diào)度的最小單位”。線程是程序中一個單一的順序控制流程。進程內(nèi)一個相對獨立的、可調(diào)度的執(zhí)行單元，是系統(tǒng)獨立調(diào)度和分派CPU的基本單位指運行中的程序的調(diào)度單位。在單個程序中同時運行多個線程完成不同的工作，稱為多線程。

進程和線程區(qū)別

進程是資源分配的基本單位。所有與該進程有關的資源，都被記錄在進程控制塊PCB中。以表示該進程擁有這些資源或正在使用它們。另外，進程也是搶占處理機的調(diào)度單位，它擁有一個完整的虛擬地址空間。當進程發(fā)生調(diào)度時，不同的進程擁有不同的虛擬地址空間，而同一進程內(nèi)的不同線程共享同一地址空間。 與進程相對應，線程與資源分配無關，它屬于某一個進程，并與進程內(nèi)的其他線程一起共享進程的資源。線程只由相關堆棧（系統(tǒng)棧或用戶棧）寄存器和線程控制表TCB組成。寄存器可被用來存儲線程內(nèi)的局部變量，但不能存儲其他線程的相關變量。 通常在一個進程中可以包含若干個線程，它們可以利用進程所擁有的資源。在引入線程的操作系統(tǒng)中，通常都是把進程作為分配資源的基本單位，而把線程作為獨立運行和獨立調(diào)度的基本單位。 由于線程比進程更小，基本上不擁有系統(tǒng)資源，故對它的調(diào)度所付出的開銷就會小得多，能更高效的提高系統(tǒng)內(nèi)多個程序間并發(fā)執(zhí)行的程度，從而顯著提高系統(tǒng)資源的利用率和吞吐量。 因而近年來推出的通用操作系統(tǒng)都引入了線程，以便進一步提高系統(tǒng)的并發(fā)性，并把它視為現(xiàn)代操作系統(tǒng)的一個重要指標。

線程與進程的區(qū)別可以歸納為以下4點：

• 地址空間和其它資源（如打開文件）：進程間相互獨立，同一進程的各線程間共享。某進程內(nèi)的線程在其它進程不可見。

• 通信：進程間通信IPC，線程間可以直接讀寫進程數(shù)據(jù)段（如全局變量）來進行通信——需要進程同步和互斥手段的輔助，以保證數(shù)據(jù)的一致性。

• 調(diào)度和切換：線程上下文切換比進程上下文切換要快得多。

• 在多線程OS中，進程不是一個可執(zhí)行的實體。

多進程和多線程的比較

對比維度	多進程	多線程	總結
數(shù)據(jù)共享、同步	數(shù)據(jù)共享復雜，同步簡單	數(shù)據(jù)共享簡單，同步復雜	各有優(yōu)劣
內(nèi)存、CPU	占用內(nèi)存多，切換復雜，CPU利用率低	占用內(nèi)存少，切換簡單，CPU利用率高	線程占優(yōu)
創(chuàng)建、銷毀、切換	復雜，速度慢	簡單，速度快	線程占優(yōu)
編程、調(diào)試	編程簡單，調(diào)試簡單	編程復雜，調(diào)試復雜	進程占優(yōu)
可靠性	進程間不會互相影響	一個線程掛掉將導致整個進程掛掉	進程占優(yōu)
分布式	適用于多核、多機，擴展到多臺機器簡單	適合于多核	進程占優(yōu)

總結，進程和線程還可以類比為火車和車廂：

• 線程在進程下行進（單純的車廂無法運行）
• 一個進程可以包含多個線程（一輛火車可以有多個車廂）
• 不同進程間數(shù)據(jù)很難共享（一輛火車上的乘客很難換到另外一輛火車，比如站點換乘）
• 同一進程下不同線程間數(shù)據(jù)很易共享（A車廂換到B車廂很容易）
• 進程要比線程消耗更多的計算機資源（采用多列火車相比多個車廂更耗資源）
• 進程間不會相互影響，一個線程掛掉將導致整個進程掛掉（一列火車不會影響到另外一列火車，但是如果一列火車上中間的一節(jié)車廂著火了，將影響到該趟火車的所有車廂）
• 進程可以拓展到多機，進程最多適合多核（不同火車可以開在多個軌道上，同一火車的車廂不能在行進的不同的軌道上）
• 進程使用的內(nèi)存地址可以上鎖，即一個線程使用某些共享內(nèi)存時，其他線程必須等它結束，才能使用這一塊內(nèi)存。（比如火車上的洗手間）－”互斥鎖（mutex）”
• 進程使用的內(nèi)存地址可以限定使用量（比如火車上的餐廳，最多只允許多少人進入，如果滿了需要在門口等，等有人出來了才能進去）－“信號量（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又是什么呢？來看看官方的解釋：

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.

Python代碼的執(zhí)行由Python 虛擬機(也叫解釋器主循環(huán)，CPython版本)來控制，Python 在設計之初就考慮到要在解釋器的主循環(huán)中，同時只有一個線程在執(zhí)行，即在任意時刻，只有一個線程在解釋器中運行。對Python 虛擬機的訪問由全局解釋器鎖（GIL）來控制，正是這個鎖能保證同一時刻只有一個線程在運行。
GIL 有什么好處？簡單來說，它在單線程的情況更快，并且在和 C 庫結合時更方便，而且不用考慮線程安全問題，這也是早期 Python 最常見的應用場景和優(yōu)勢。另外，GIL的設計簡化了CPython的實現(xiàn)，使得對象模型，包括關鍵的內(nèi)建類型如字典，都是隱含可以并發(fā)訪問的。鎖住全局解釋器使得比較容易的實現(xiàn)對多線程的支持，但也損失了多處理器主機的并行計算能力。

在多線程環(huán)境中，Python 虛擬機按以下方式執(zhí)行：

1. 設置GIL
2. 切換到一個線程去運行
3. 運行直至指定數(shù)量的字節(jié)碼指令，或者線程主動讓出控制（可以調(diào)用sleep(0)）
4. 把線程設置為睡眠狀態(tài)
5. 解鎖GIL

6. 再次重復以上所有步驟

Python3.2前，GIL的釋放邏輯是當前線程遇見IO操作或者ticks計數(shù)達到100（ticks可以看作是python自身的一個計數(shù)器，專門做用于GIL，每次釋放后歸零，這個計數(shù)可以通過 sys.setcheckinterval 來調(diào)整），進行釋放。因為計算密集型線程在釋放GIL之后又會立即去申請GIL，并且通常在其它線程還沒有調(diào)度完之前它就已經(jīng)重新獲取到了GIL，就會導致一旦計算密集型線程獲得了GIL，那么它在很長一段時間內(nèi)都將占據(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上被喚醒后的線程會醒著等待到切換時間后又進入待調(diào)度狀態(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程序直接調(diào)用C語言編譯的動態(tài)鏈接庫的導出函數(shù)。（with nogil調(diào)出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)調(diào)用，它非常特殊。普通的函數(shù)，調(diào)用一次，返回一次，但是fork()調(diào)用一次，返回兩次，因為操作系統(tǒng)自動把當前進程（父進程）復制了一份（子進程），然后，分別在父進程和子進程內(nèi)返回。子進程永遠返回0，而父進程返回子進程的ID。這樣做的理由是，一個父進程可以fork出很多子進程，所以，父進程要記下每個子進程的ID，而子進程只需要調(diào)用getpid()就可以拿到父進程的ID。 Python的os模塊封裝了常見的系統(tǒng)調(diào)用，其中就包括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.

      
    

有了fork調(diào)用，一個進程在接到新任務時就可以復制出一個子進程來處理新任務，常見的Apache服務器就是由父進程監(jiān)聽端口，每當有新的http請求時，就fork出子進程來處理新的http請求。 由于Windows沒有fork調(diào)用，上面的代碼在Windows上無法運行。由于Python是跨平臺的，自然也應該提供一個跨平臺的多進程支持。multiprocessing模塊就是跨平臺版本的多進程模塊。multiprocessing模塊封裝了fork()調(diào)用，使我們不需要關注fork()的細節(jié)。由于Windows沒有fork調(diào)用，因此，multiprocessing需要“模擬”出fork的效果。

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：表示調(diào)用對象，即子進程要執(zhí)行的任務，你可以傳入方法名
• name：為子進程設定名稱
• args：要傳給target函數(shù)的位置參數(shù)，以元組方式進行傳入。
• kwargs：要傳給target函數(shù)的字典參數(shù)，以字典方式進行傳入。

實例方法：

• start()：啟動進程，并調(diào)用該子進程中的p.run()
• run()：進程啟動時運行的方法，正是它去調(diào)用target指定的函數(shù)，我們自定義類的類中一定要實現(xiàn)該方法
• terminate()：強制終止進程p，不會進行任何清理操作，如果p創(chuàng)建了子進程，該子進程就成了僵尸進程，使用該方法需要特別小心這種情況。如果p還保存了一個鎖那么也將不會被釋放，進而導致死鎖
• is_alive()：返回進程是否在運行。如果p仍然運行，返回True
• join([timeout])：進程同步，主進程等待子進程完成后再執(zhí)行后面的代碼。線程等待p終止（強調(diào)：是主線程處于等的狀態(tài)，而p是處于運行的狀態(tài)）。timeout是可選的超時時間（超過這個時間，父線程不再等待子線程，繼續(xù)往下執(zhí)行），需要強調(diào)的是，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ù)量的進程，供用戶調(diào)用，當有新的請求提交到Pool中時，如果池還沒有滿，那么就會創(chuàng)建一個新的進程用來執(zhí)行該請求；但如果池中的進程數(shù)已經(jīng)達到規(guī)定最大值，那么該請求就會等待，直到池中有進程結束，就重用進程池中的進程。 構造方法：Pool([processes[, initializer[, initargs[, maxtasksperchild[, context]]]]])

• processes ：要創(chuàng)建的進程數(shù)，如果省略，將默認使用cpu_count()返回的數(shù)量。
• initializer：每個工作進程啟動時要執(zhí)行的可調(diào)用對象，默認為None。如果initializer是None，那么每一個工作進程在開始的時候會調(diào)用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),然后返回結果。需要強調(diào)的是：此操作并不會在所有池工作進程中并執(zhí)行func函數(shù)。如果要通過不同參數(shù)并發(fā)地執(zhí)行func函數(shù)，必須從不同線程調(diào)用p.apply()函數(shù)或者使用p.apply_async()。它是阻塞的。apply很少使用
• apply_async(func[, arg[, kwds={}[, callback=None]]])：在一個池工作進程中執(zhí)行func(args,*kwargs),然后返回結果。此方法的結果是AsyncResult類的實例，callback是可調(diào)用對象，接收輸入?yún)?shù)。當func的結果變?yōu)榭捎脮r，將理解傳遞給callback。callback禁止執(zhí)行任何阻塞操作，否則將接收其他異步操作中的結果。它是非阻塞。
• map(func, iterable[, chunksize=None])：Pool類中的map方法，與內(nèi)置的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()之后調(diào)用，讓其不再接受新的Process。
• terminate()：結束工作進程，不再處理未處理的任務。

方法apply_async()和map_async()的返回值是AsyncResul的實例obj。實例具有以下方法：

• get()：返回結果，如果有必要則等待結果到達。timeout是可選的。如果在指定時間內(nèi)還沒有到達，將引發(fā)異常。如果遠程操作中引發(fā)了異常，它將在調(diào)用此方法時再次被引發(fā)。
• ready()：如果調(diào)用完成，返回True
• successful()：如果調(diào)用完成且沒有引發(fā)異常，返回True，如果在結果就緒之前調(diào)用此方法，引發(fā)異常
• wait([timeout])：等待結果變?yōu)榭捎谩?/span>
• terminate()：立即終止所有工作進程，同時不執(zhí)行任何清理或結束任何掛起工作。如果p被垃圾回收，將自動調(diào)用此函數(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)內(nèi)部執(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為正值，那么在等待時間內(nèi)沒有取到任何元素，會拋出Queue.Empty異常。如果blocked為False，有兩種情況存在，如果Queue有一個值可用，則立即返回該值，否則，如果隊列為空，則立即拋出Queue.Empty異常。若不希望在empty的時候拋出異常，令blocked為True或者參數(shù)全部置空即可。
• get_nowait()：同q.get(False)
• put_nowait()：同q.put(False)
• empty()：調(diào)用此方法時q為空則返回True，該結果不可靠，比如在返回True的過程中，如果隊列中又加入了項目。
• full()：調(diào)用此方法時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)，無法等待其結束，只能強行終止

      
    

JoinableQueue就像是一個Queue對象，但隊列允許項目的使用者通知生成者項目已經(jīng)被成功處理。通知進程是使用共享的信號和條件變量來實現(xiàn)的。 構造方法：JoinableQueue([maxsize])

• maxsize：隊列中允許最大項數(shù)，省略則無大小限制。

實例方法 JoinableQueue的實例p除了與Queue對象相同的方法之外還具有：

• task_done()：使用者使用此方法發(fā)出信號，表示q.get()的返回項目已經(jīng)被處理。如果調(diào)用此方法的次數(shù)大于從隊列中刪除項目的數(shù)量，將引發(fā)ValueError異常
• join():生產(chǎn)者調(diào)用此方法進行阻塞，直到隊列中所有的項目均被處理。阻塞將持續(xù)到隊列中的每個項目均調(diào)用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內(nèi)調(diào)用q.join保證了consumer已經(jīng)處理完隊列中的所有元素

    p.start()

    producer(seq, q)

    print('主線程')

      
    

Value，Array（用于進程通信，資源共享）

multiprocessing 中Value和Array的實現(xiàn)原理都是在共享內(nèi)存中創(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的實例就不會被鎖保護，它將不是進程安全的。

typecode_or_type支持的類型：

      
        | 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表示管道兩端的連接對象，強調(diào)一點：必須在產(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被垃圾回收，將自動調(diào)用此方法
• 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ā)出，然后調(diào)用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()：關閉管理器對象，只能在調(diào)用了start()方法之后調(diào)用

實例屬性：

• address：只讀屬性，管理器服務器正在使用的地址

SyncManager，以下類型均不是進程安全的，需要加鎖.. 實例方法：

• 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的線程可以再次調(diào)用acquire()，釋放鎖時需要調(diào)用release()相同次數(shù)。可以認為RLock包含一個鎖定池和一個初始值為0的計數(shù)器，每次成功調(diào)用 acquire()/release()，計數(shù)器將+1/-1，為0時鎖處于未鎖定狀態(tài)。

構造方法：RLock()

實例方法：

• acquire([timeout])：同Lock
• release(): 同Lock

Semaphore（信號量）

信號量是一個更高級的鎖機制。信號量內(nèi)部有一個計數(shù)器而不像鎖對象內(nèi)部有鎖標識，而且只有當占用信號量的線程數(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在內(nèi)部維護一個鎖對象（默認是RLock），可以在創(chuàng)建Condigtion對象的時候把瑣對象作為參數(shù)傳入。Condition也提供了acquire, release方法，其含義與鎖的acquire, release方法一致，其實它只是簡單的調(diào)用內(nèi)部鎖對象的對應的方法而已。Condition還提供了其他的一些方法。

構造方法：Condition([lock/rlock])

• 可以傳遞一個Lock/RLock實例給構造方法，否則它將自己生成一個RLock實例。

實例方法：

• acquire([timeout])：首先進行acquire，然后判斷一些條件。如果條件不滿足則wait
• release()：釋放 Lock
• wait([timeout]): 調(diào)用這個方法將使線程進入Condition的等待池等待通知，并釋放鎖。使用前線程必須已獲得鎖定，否則將拋出異常。處于wait狀態(tài)的線程接到通知后會重新判斷條件。
• notify(): 調(diào)用這個方法將從等待池挑選一個線程并通知，收到通知的線程將自動調(diào)用acquire()嘗試獲得鎖定（進入鎖定池）；其他線程仍然在等待池中。調(diào)用這個方法不會釋放鎖定。使用前線程必須已獲得鎖定，否則將拋出異常。
• notifyAll(): 調(diào)用這個方法將通知等待池中所有的線程，這些線程都將進入鎖定池嘗試獲得鎖定。調(diào)用這個方法不會釋放鎖定。使用前線程必須已獲得鎖定，否則將拋出異常。

使用示例：

      
        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內(nèi)部包含了一個標志位，初始的時候為false。可以使用set()來將其設置為true；或者使用clear()將其從新設置為false；可以使用is_set()來檢查標志位的狀態(tài)；另一個最重要的函數(shù)就是wait(timeout=None)，用來阻塞當前線程，直到event的內(nèi)部標志位被設置為true或者timeout超時。如果內(nèi)部標志位為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')

      
    

其他內(nèi)容

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í)行異步調(diào)用。

      
        class concurrent.futures.ThreadPoolExecutor(max_workers)

      
    

使用max_workers數(shù)目的線程池執(zhí)行異步調(diào)用。

ProcessPoolExecutor對象

ThreadPoolExecutor類是Executor子類，使用進程池執(zhí)行異步調(diào)用。

      
        class concurrent.futures.ProcessPoolExecutor(max_workers=None)

      
    

使用max_workers數(shù)目的進程池執(zhí)行異步調(diào)用，如果max_workers為None則使用機器的處理器數(shù)目（如4核機器max_worker配置為None時，則使用4個進程進行異步并發(fā)）。

submit()方法

Executor中定義了submit()方法，這個方法的作用是提交一個可執(zhí)行的回調(diào)task，并返回一個future實例。future對象代表的就是給定的調(diào)用。

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)迭代器，迭代器中的回調(diào)執(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()等異步操作后調(diào)用。使用with語句可以避免顯式調(diào)用此方法。

Executor.shutdown(wait=True)

Future

Future可以理解為一個在未來完成的操作，這是異步編程的基礎。通常情況下，我們執(zhí)行io操作，訪問url時（如下）在等待結果返回之前會產(chǎn)生阻塞，cpu不能做其他事情，而Future的引入幫助我們在等待的這段時間可以完成其他的操作。 Future類封裝了可調(diào)用的異步執(zhí)行。Future 實例通過 Executor.submit()方法創(chuàng)建。

• cancel()：試圖取消調(diào)用。如果調(diào)用當前正在執(zhí)行，并且不能被取消，那么該方法將返回False，否則調(diào)用將被取消，方法將返回True。
• cancelled()：如果成功取消調(diào)用，返回True。
• running()：如果調(diào)用當前正在執(zhí)行并且不能被取消，返回True。
• done()：如果調(diào)用成功地取消或結束了，返回True。
• result(timeout=None)：返回調(diào)用返回的值。如果調(diào)用還沒有完成，那么這個方法將等待超時秒。如果調(diào)用在超時秒內(nèi)沒有完成，那么就會有一個Futures.TimeoutError將報出。timeout可以是一個整形或者浮點型數(shù)值，如果timeout不指定或者為None,等待時間無限。如果futures在完成之前被取消了，那么 CancelledError 將會報出。
• exception(timeout=None)：返回調(diào)用拋出的異常，如果調(diào)用還未完成，該方法會等待timeout指定的時長，如果該時長后調(diào)用還未完成，就會報出超時錯誤futures.TimeoutError。timeout可以是一個整形或者浮點型數(shù)值，如果timeout不指定或者為None,等待時間無限。如果futures在完成之前被取消了，那么 CancelledError 將會報出。如果調(diào)用完成并且無異常報出，返回None.
• add_done_callback(fn)：將可調(diào)用fn捆綁到future上，當Future被取消或者結束運行，fn作為future的唯一參數(shù)將會被調(diào)用。如果future已經(jīng)運行完成或者取消，fn將會被立即調(diào)用。
• 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   作者：錢魏Way  來源： https://www.biaodianfu.com/python-multi-thread-and-multi-process.html

72張PNG，圖解機器學習

PyTorch從入門到項目實踐

機器學習算法與Python實戰(zhàn)P7：概率論基礎（下）

PyTorch那些事兒（二）：PyTorch建模示例

PyTorch那些事兒（十三）：深度學習模型訓練過程

PyTorch那些事兒（十四）：深度學習訓練過程可視化

可能是全網(wǎng)最全的速查表：Python機器學習ChatGPT線性代數(shù)微積分概率統(tǒng)計