前言

本文的文字及圖片來源於網路,僅供學習、交流使用,不具有任何商業用途,版權歸原作者所有,如有問題請及時聯絡我們以作處理。

作者：東哥IT筆記

現在很多CPU都支援多核，甚至是手機都已經開始支援多核了。而Python的GIL（Global Interpreter Locko）則使得其沒法使用這些多核帶來的優勢。還好從Python2.6開始，引入了multiprocessing模組，我們終於可以使用多核帶來的便利了。

本文，你會學習到下面這些內容：

• 使用多程式的優點
• 使用多程式的缺點
• 使用multiprocessing來建立多程式
• Process的子類化
• 建立程式池

本文並不是一個multiprocessing的全面的介紹，假如你想全面的瞭解它，可以參見官方的檔案：

https://docs.python.org/2/library/multiprocessing.html

下面讓我們開始吧！

使用多程式的優點

使用多程式有很多優點：

• 多程式使用獨立的記憶體空間
• 相比於執行緒，程式碼更加直觀
• 能夠使用多個CPU/多核
• 避免GIL
• 子程式可以被kill（和thread不同）
• multiprocessing有和threading.Thread類似的介面
• 對CPU繫結的程式比較好（加密，二進位制搜尋，矩陣乘法等）

下面我們來看看使用多程式有什麼缺點：

使用多程式的缺點

使用多程式也有一些缺點：

• 程式間通訊更加複雜
• 記憶體的佔用大於執行緒

使用multiprocessing來建立程式

multiprocessing是用來模擬threading.Thread類工作的。下面就是一個使用它的例子：

import multiprocessing

import random

import time

def worker(name: str) -> None:

    print(f'Started worker {name}')

    worker_time = random.choice(range(1, 5))

    time.sleep(worker_time)

    print(f'{name} worker finished in {worker_time} seconds')

if __name__ == '__main__':
 

    processes = []

    for i in range(5):

        process = multiprocessing.Process(target=worker,

                                          args=(f'computer_{i}',))

        processes.append(process)

        process.start()

    for proc in processes:

        proc.join()

首先第一步需要import multiprocessing模組，另外兩個import分別是為random和time服務的。

worker函式就是用來假裝做一些事情，傳入一個name的引數，沒有什麼返回，他首先列印name的值，然後隨機sleep一段時間用來模擬做一段很長時間的工作，最後列印work finish。

緊接著，你使用multiprocessing.Process建立了5個程式，他的使用和threading.Tread()比較類似，你告訴Process哪個目標函式需要呼叫，以及會傳入什麼引數給他們，然後你呼叫了start函式來啟動程式。另外你會把這些程式加入到一個list中。

最後，你遍歷這個list，呼叫join方法，這個方法其實就是告訴Python等到程式結束。

假如你run這個函式，你會看到類似下面這樣的輸出：

其實你每次執行這個函式，結果都會有稍許的不同，主要還是因為你呼叫了random函式，你可以試試，看看你自己的輸出。

Process的子類化

multiporcessing模組中的Process類是可以子類化的，他和threading.thread的類差不多。我們來看下面程式碼：

# worker_thread_subclass.py

import random

import multiprocessing

import time

class WorkerProcess(multiprocessing.Process):

    def __init__(self, name):

        multiprocessing.Process.__init__(self)

        self.name = name

    def run(self):

        """

        Run the thread

        """

        worker(self.name)

def worker(name: str) -> None:

    print(f'Started worker {name}')

    worker_time = random.choice(range(1, 5))

    time.sleep(worker_time)

    print(f'{name} worker finished in {worker_time} seconds')

if __name__ == '__main__':

    processes = []

    for i in range(5):

        process = WorkerProcess(name=f'computer_{i}')

        processes.append(process)

        process.start()

    for process in processes:

        process.join()

這裡，我們寫了一個multiprocess.Process()的子類，並且重寫了run()方法。

其他方面和上面的例子其實是類似的，現在我們可以來看看具體的輸出，和上面的也類似。

建立一個程式池

假如你有很多程式需要執行，有時你希望能夠限制程式執行的數目。比如說，你需要執行20個程式，但是你只有4個核，那麼你可以使用multiprocessing模組來建立一個程式池，用它來限制每次程式執行的數目到4個。

下面是示例的程式碼：

import random

import time

from multiprocessing import Pool

def worker(name: str) -> None:

    print(f'Started worker {name}')

    worker_time = random.choice(range(1, 5))

    time.sleep(worker_time)

    print(f'{name} worker finished in {worker_time} seconds')

if __name__ == '__main__':

    process_names = [f'computer_{i}' for i in range(15)]

    pool = Pool(processes=5)

    pool.map(worker, process_names)

    pool.terminate()

這個例子中，worker函式還是一樣的，主要是後面的程式碼， 我們建立了一個程式池，它的數目是5，也就意味著最大的執行數目是5。使用這個pool，你需要呼叫map()方法，然後把你需要的呼叫的方法和引數傳遞給他。

這樣的話，Python每次只會使用5個程式來執行直到結束。最後你需要呼叫terminate()函式，否則你會發現下面的錯誤：

/Library/Frameworks/Python.framework/Versions/3.8/lib/python3.8/multiprocessing/resource_tracker.py:216:

UserWarning: resource_tracker: There appear to be 6 leaked semaphore objects to clean up at shutdown

這個程式碼的具體輸出如下所示：