Python3 多進程詳解

一、多進程概述

多進程是指同一時刻有多個進程同時執行，每個進程都有獨立的內存空間、寄存器和程序計數器等。多進程可同時利用多個CPU進行並行處理，提高計算機的執行效率。

在 Python 中，使用 multiprocessing 模塊可以輕鬆實現多進程操作。

二、創建進程

Python 中創建子進程的方式有多種，這裡主要介紹兩種方式：使用 Process 對象和使用 Pool 對象。

1. 使用 Process 對象

Process 對象用於表示進程，它有一個構造函數，可以傳入目標函數和參數，用於啟動子進程。

import multiprocessing

def worker(arg):
    print(f'Process {arg} is running.')

if __name__ == '__main__':
    for i in range(4):
        p = multiprocessing.Process(target=worker, args=(i,))
        p.start()

上面的代碼中，創建了 4 個子進程，並分別傳入參數 0、1、2、3，子進程將執行 worker 函數列印輸出對應的進程號。

需要注意的是，在 Windows 平台下，由於沒有 fork 調用，所以 multiprocessing 模塊需要在 __name__ == ‘__main__’ 的保護下運行，以防止無限遞歸。

2. 使用 Pool 對象

Pool 對象用於管理進程池，它其實就是維護了一個進程的列表，當需要使用的時候，就從這個池中去取，用完之後再還回去。這種方式可以提高進程的重複利用率，從而降低系統開銷。

import multiprocessing
import time

def worker(arg):
    time.sleep(1)
    print(f'Process {arg} is running.')

if __name__ == '__main__':
    with multiprocessing.Pool(processes=3) as pool:
        pool.map(worker, range(10))

上面的代碼中，使用 Pool 對象創建了一個擁有 3 個進程的進程池，然後用 map 函數調用 worker 函數，傳入一個包含 10 個元素的列表，每個元素都作為 worker 函數的參數。

可以看到，進程池會維護 3 個進程，每個進程的運行時間為 1 秒，最終輸出的信息認為是重疊在一起的結果，但實際上是 10 個進程分別執行後的結果。

三、進程間通信

由於每個進程都有自己獨立的內存空間，所以進程間通信（IPC）需要使用特定的機制進行傳遞數據。

在 Python 中，multiprocess 模塊提供了多種進程通信方式，比如 Queue、Pipe 等。

1. Queue 隊列

Queue 隊列是進程間通信的一種簡單方式，用於在多個進程之間安全地傳遞數據。

import multiprocessing

def worker(q):
    while True:
        value = q.get()
        if value is None:
            break
        print(value)

if __name__ == '__main__':
    q = multiprocessing.Queue()
    p = multiprocessing.Process(target=worker, args=(q,))
    p.start()

    for i in range(10):
        q.put(i)

    q.put(None)
    p.join()

上面的代碼中，創建了一個 Queue 對象，並以它作為參數啟動了子進程，主進程通過 put 函數向隊列中添加 10 個數字，子進程不斷從隊列中取出數據並列印，當取出 None 時，子進程停止運行。

2. Pipe 管道

Pipe 管道是另一種用於進程間通信的方式。它返回一對對象，包含兩個鏈接端，一個稱為發送端，另一個稱為接收端。

import multiprocessing

def worker(conn):
    conn.send('hello')
    print(conn.recv())
    conn.close()

if __name__ == '__main__':
    parent_conn, child_conn = multiprocessing.Pipe()
    p = multiprocessing.Process(target=worker, args=(child_conn,))
    p.start()

    print(parent_conn.recv())
    parent_conn.send('world')
    p.join()

上面的代碼中，創建了一個管道對象，並以它作為參數啟動了子進程，主進程向管道中發送字元串 ‘hello’，子進程從管道中接收到數據並列印，然後向管道中發送字元串 ‘world’，主進程從管道中接收到數據並列印，子進程退出。

四、進程池中返回值

使用 pool 對象創建進程池時，可以通過 apply_async 方法指定特定進程要執行的任務，並且可以通過 get 方法獲得進程返回的值。

import multiprocessing

def sum(a, b):
    return a + b

if __name__ == '__main__':
    with multiprocessing.Pool(processes=1) as pool:
        results = []
        result = pool.apply_async(sum, (1, 2))
        results.append(result)

        for result in results:
            print(result.get())

上面的代碼中，使用 apply_async 方法傳入函數和參數，每個進程在完成任務後會返回一個結果，使用 get 方法來得到結果。

五、進程間共享變數

Python 中的多進程間是沒有共享變數的，每個進程都有獨立的內存空間，但共享部分需要使用特定的機制來進行操作。

1. Value 變數

Value 變數是 Python 中的一種特殊數據類型，它可以在多進程間進行共享。

import multiprocessing

def change_value(value):
    value.value = 100

if __name__ == '__main__':
    num = multiprocessing.Value('d', 0.0)
    p = multiprocessing.Process(target=change_value, args=(num,))
    p.start()
    p.join()

    print(num.value)

上面的代碼中，創建了一個 Value 變數，然後以它作為參數啟動了一個子進程，子進程執行 change_value 函數，將 Value 變數的值改為 100，最終將結果輸出。

2. Array 數組

Array 數組是另一種可以在多進程間使用的數據類型。這個數據類型表示一種可以容納相同類型數據的可變序列。

import multiprocessing

def modify_array(arr):
    for i in range(len(arr)):
        arr[i] *= 2

if __name__ == '__main__':
    arr = multiprocessing.Array('i', range(10))
    p = multiprocessing.Process(target=modify_array, args=(arr,))
    p.start()
    p.join()

    print(arr[:])

上面的代碼中，創建了一個 Array 數組，然後以它作為參數啟動了一個子進程，子進程執行 modify_array 函數，將數組中的所有數字都乘以 2，最終將結果輸出。

六、進程之間的同步

在多進程的操作中，有時候需要控制進程的執行順序或者進程之間有依賴關係的時候，需要使用同步機制。

1. Lock 鎖

Lock 鎖是 Python 中的一種同步機制，它用於控制多個進程對共享資源的訪問。

import multiprocessing

def increment(value, lock):
    for idx in range(10000):
        with lock:
            value.value += 1

if __name__ == '__main__':
    lock = multiprocessing.Lock()
    value = multiprocessing.Value('i', 0)

    processes = [multiprocessing.Process(target=increment, args=(value, lock)) for i in range(10)]

    for process in processes:
        process.start()
    for process in processes:
        process.join()

    print(value.value)

上面的代碼中，創建了一個 Lock 對象，以 Value 變數和 Lock 對象作為參數啟動了 10 個子進程，每個子進程都會將 Value 變數的值加 1，執行鎖定，保證多個進程對變數的訪問是互斥的。

2. Semaphore 信號量

Semaphore 信號量是 Python 中的另一種同步機制，它可以控制多個進程對共享資源的訪問，並設置鎖定次數。

import multiprocessing
import time

def worker(s, i):
    with s:
        print(f'Worker {i} start')
        time.sleep(1)
        print(f'Worker {i} end')

if __name__ == '__main__':
    s = multiprocessing.Semaphore(2)

    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(s, i))
        p.start()

上面的代碼中，創建了一個 Semaphore 對象，以 Semaphore 對象和進程編號作為參數啟動 5 個子進程，每個子進程獲得鎖之後列印相應的信息，然後等待 1 秒再釋放鎖。

七、多進程的其他注意事項

在 Python 中，多進程操作有一些需要注意的問題。

1. 進程的資源限制

在 Python 進行多進程操作中，需要注意操作系統的資源限制。有些系統的資源分配是靜態的，需要在運行時限制進程數量。在 Linux 中，可以使用 ulimit 命令查看系統資源。

2. 多進程的數據傳遞問題

在 Python 進行多進程操作時，需要注意數據的傳遞問題。由於每個進程都有自己獨立的內存空間，因此需要使用特定的機制進行數據的傳遞。

3. 進程的生命周期

在 Python 進行多進程操作時，需要注意進程的生命周期問題。進程的生命周期由操作系統決定，必須等待進程執行完畢才能退出。

4. 進程間的信號處理

在 Python 進行多進程操作時，需要注意進程間的信號處理問題。不同進程之間的信號處理是相互獨立的，需要注意在適當的時候的發送和接收信號。

八、總結

在 Python 中，多進程操作是一個非常強大的功能，能夠大大提高程序的執行效率。本文對多進程的概念、創建、進程間通信、進程池中返回值、進程間共享變數、進程之間的同步、以及多進程操作中需要注意的一些問題進行了詳細的介紹和講解。