介紹如何使用 Python 的 multiprocessing 模組，開發多核心平行運算程式。
Python 的
multiprocessing 是一個多行程模組，功能類似多執行緒的 threading 模組，可以讓開發者使用多核心的 CPU 進行平行化程式設計，加速程式處理速度。
Pool 資源池
Pool 是由多個工作者行程（worker process）所組成的資源池，可以將工作分配給其內部的各個工作者行程進行平行處理。
map 函數
Pool 最簡單的使用方式就是將資料以列表的形式打包好，透過其 map 函數將實際的工作處理函數與列表資料傳入其中。
假設我們想要以 Pool 的 map 函數平行計算多個數值的平方，就可以這樣寫：
from multiprocessing import Pool # 計算數值平方的函數 def f(x): return x*x # 建立含有 4 個工作者行程的 Pool with Pool(processes=4) as p: # 以 map 平行計算數值的平方 print(p.map(f, [1, 2, 3]))
Pool 的 map 函數可以視為 Python 內建 map 函數的平行化版本，呼叫 Pool 的 map 函數可以將資料分割之後，再以 Pool 中的工作者行程平行處理，等到所有資料都處理完成之後，再一次傳回所有的結果。
Pool 的 map 函數有一個 chunksize 參數可以調整資料分割的大小，當資料列表的長度很長的時候，可以將 chunksize 設為較大的值，讓 map 每次切多一點資料給各個工作者行程，這樣可以減少資料傳遞的次數，提升程式執行效能。以下是比較不同 chunksize 對程式執行速度的範例程式：
from multiprocessing import Pool import time def f(x): return x*x with Pool(processes=4) as p: start = time.time() # 開始測量執行時間 # 設定 chunksize 為 1 result1 = p.map(f, range(1000000), chunksize=1) end = time.time() # 結束測量執行時間 print("chunksize 為 1，執行時間為 %f 秒" % (end - start)) start = time.time() # 開始測量執行時間 # 設定 chunksize 為 1000 result2 = p.map(f, range(1000000), chunksize=1000) end = time.time() # 結束測量執行時間 print("chunksize 為 1000，執行時間為 %f 秒" % (end - start)) if result1 == result2: print("結果相同")
chunksize 為 1，執行時間為 24.855215 秒 chunksize 為 1000，執行時間為 0.257618 秒 結果相同
調整 chunksize 並不會影響程式結構或計算結果，但是卻對執行時間影響很大。
imap 函數
imap 函數是 map 函數的可迭代（iterable）版本，其參數的用法皆與 map 函數相同：
from multiprocessing import Pool def f(x): return x*x with Pool(processes=4) as p: # 以 imap 平行計算數值的平方 for i in p.imap(f, [1, 2, 3]): print(i)
1 4 9
imap 的 chunksize 設定方式與概念皆與 map 相同：
from multiprocessing import Pool def f(x): return x*x with Pool(processes=4) as p: # 設定 chunksize 為 1000 for i in p.imap(f, range(1000000), chunksize=1000): print(i)
imap_unordered 函數
imap_unordered 函數跟 imap 函數非常類似，只不過 imap_unordered 所傳回的計算結果不會完全依照輸入資料的順序傳回，而是將先處理完成的資料先傳回，可增進整體程式的執行效率：
from multiprocessing import Pool def f(x): return x*x with Pool(processes=4) as p: # 以 imap_unordered 平行計算數值的平方 for i in p.imap_unordered(f, [1, 2, 3]): print(i)
1 9 4
imap_unordered 的 chunksize 設定方式也都相同。
以下是一個用來比較 imap_unordered 與 imap 差異的範例，imap_unordered 會將先處理完成的資料先傳回，而 imap 則是為了要保持原有的資料順序，所以會有潛在的等待時間，這就會造成程式執行效率上的差異：
from multiprocessing import Pool import time def f(x): time.sleep(x) with Pool(processes=4) as p: start = time.time() # 開始測量執行時間 # 以 imap 平行處理 for i in p.imap(f, [3, 2, 1]): time.sleep(1) end = time.time() # 結束測量執行時間 print("imap 執行時間為 %f 秒" % (end - start)) start = time.time() # 開始測量執行時間 # 以 imap_unordered 平行處理 for i in p.imap_unordered(f, [3, 2, 1]): time.sleep(1) end = time.time() # 結束測量執行時間 print("imap_unordered 執行時間為 %f 秒" % (end - start))
imap 執行時間為 6.007628 秒 imap_unordered 執行時間為 4.004911 秒
所以如果我們不在意計算結果的傳回順序，就可以採用效率較好的 imap_unordered 函數。
starmap 函數
starmap 函數可以用來處理多輸入參數的狀況，以下是每次傳遞兩個輸入的數值，平行計算總合的範例。
from multiprocessing import Pool # 計算總和 def f(x, y): return x+y with Pool(processes=4) as p: # 以 starmap 傳遞多個輸入參數資料 print(p.starmap(f, [(1, 2), (3, 4)]))
[3, 7]
apply_async 函數
apply_async 函數可以將指定的函數放進 Pool 資源池中執行，以下是平行執行多個函數的範例：
from multiprocessing import Pool import time def f1(x): time.sleep(1) return x*x def f2(x): time.sleep(1) return x*x*x with Pool(processes=4) as p: start = time.time() # 開始測量執行時間 # 平行呼叫多個函數 r1 = p.apply_async(f1, (2,)) r2 = p.apply_async(f2, (2,)) # 取得計算結果 print(r1.get(timeout=3)) print(r2.get(timeout=3)) end = time.time() # 結束測量執行時間 print("執行時間：%f 秒" % (end - start))
4 8 執行時間：1.002772 秒
map_async 函數
map_async 函數是 map 函數的非同步版本，可以將工作放進 Pool 資源池中執行，等執行完成後再一次取回：
from multiprocessing import Pool import time def f1(x): time.sleep(1) return x*x def f2(x): time.sleep(1) return x*x*x with Pool(processes=4) as p: start = time.time() # 開始測量執行時間 # 平行呼叫多個函數 r1 = p.map_async(f1, [1, 2, 3]) r2 = p.map_async(f2, [1, 2, 3]) # 取得計算結果 print(r1.get(timeout=3)) print(r2.get(timeout=3)) end = time.time() # 結束測量執行時間 print("執行時間：%f 秒" % (end - start))
[1, 4, 9] [1, 8, 27] 執行時間：2.003540 秒
starmap_async 函數
starmap_async 函數是 starmap 函數的非同步版本，可以將工作放進 Pool 資源池中執行，等執行完成後再一次取回：
from multiprocessing import Pool import time def f1(x, y): time.sleep(1) return x+y def f2(x, y): time.sleep(1) return x*y with Pool(processes=4) as p: start = time.time() # 開始測量執行時間 # 平行呼叫多個函數 r1 = p.starmap_async(f1, [(1, 2), (3, 4)]) r2 = p.starmap_async(f2, [(1, 2), (3, 4)]) # 取得計算結果 print(r1.get(timeout=3)) print(r2.get(timeout=3)) end = time.time() # 結束測量執行時間 print("執行時間：%f 秒" % (end - start))
[3, 7] [2, 12] 執行時間：1.002267 秒
Process 行程
multiprocessing 模組中的 Process 可以用來建立獨立的行程，平行處理不同的工作。
from multiprocessing import Process # 實際要執行的工作 def f(name): print('Hello', name) # 建立獨立行程 p = Process(target=f, args=('Bob',)) # 開始執行 p.start() # 原行程可處理其他工作... # 等待獨立行程結束 p.join()
Hello Bob
共享記憶體
如果需要在行程之間共享資料，可以使用 Value 或 Array 來儲存資料，讓多個行程可以同時存取。以下是透過 Value 與 Array 取回計算結果的範例。
from multiprocessing import Process, Value, Array def f(n, a): n.value = 1.24 for i in range(len(a)): a[i] = a[i] * 2 # 建立共享記憶體的 Value 與 Array 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[:])
1.24 [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]
如果有多個行程都會同時更改到共享記憶體中的資料，就必須搭配 Lock 鎖定記憶體中的資料之後再進行修改，才不會因為競爭條件（race condition）造成資料錯誤：
from multiprocessing import Process, Value def f(c): for i in range(1000): # 取得 Lock 之後更改內容 with counter.get_lock(): c.value += 1 # 建立共享記憶體的 Value counter = Value('i', 0) # 以獨立行程執行 p1 = Process(target=f, args=(counter,)) p2 = Process(target=f, args=(counter,)) p1.start() p2.start() p1.join() p2.join() # 取得計算結果 print(counter.value)
2000
Queue
Queue 是另外一種在行程之間遞送資料的方法，以下是一個簡單的使用範例。
from multiprocessing import Process, Queue def f(q): q.put([42, None, 'hello']) # 建立共享佇列 q = Queue() p = Process(target=f, args=(q,)) p.start() # 從佇列取得資料 print(q.get()) p.join()
[42, None, 'hello']
Pipe
Pipe 可以讓兩個行程之間進行雙向的資料傳輸，以下是一個簡單的範例。
from multiprocessing import Process, Pipe def f(conn): # 透過 Pipe 傳送資料 conn.send([42, None, 'hello']) # 關閉 Pipe 連線 conn.close() # 建立可雙向傳輸的 Pipe parent_conn, child_conn = Pipe() p = Process(target=f, args=(child_conn,)) p.start() # 從 Pipe 取得資料 print(parent_conn.recv()) p.join()
[42, None, 'hello']
在預設的情況下，建立 Pipe 的時候所產生兩個 Connection 物件都具有發送與接收資料的能力（send 與 recv 函數）。