您现在的位置：程式碼之家 >> 編程語言 > >> 更多程式語言 >> 更多程式設計知識

Python multiprocessing 多進程完整指南：Process、Pool、共享記憶體與進程池最佳實務

編輯：更多程式設計知識

`multiprocessing` 是 Python 繞過 GIL (全域解譯器鎖) 實現真正並行計算的標準庫。本文系統性解析 `Process`、`Pool`、共享記憶體、進程間通訊與生產級進程池管理模式。

基礎：Process 類別


from multiprocessing import Process
import os
import time

def worker(name, delay):
    print(f"[PID:{os.getpid()}] 工作者 {name} 開始")
    time.sleep(delay)
    print(f"[PID:{os.getpid()}] 工作者 {name} 結束")

if __name__ == "__main__":  # Windows 必須
    processes = []
    for i in range(3):
        p = Process(target=worker, args=(f"W{i}", 1 + i * 0.5))
        p.start()
        processes.append(p)
    
    for p in processes:
        p.join()  # 等待結束
    
    print("所有進程完成")

繼承 Process 類別 (物件導向風格)


from multiprocessing import Process
import time

class TaskProcess(Process):
    def __init__(self, task_id, workload):
        super().__init__()
        self.task_id = task_id
        self.workload = workload
        self.result = None
    
    def run(self):
        """進程進入點，不可直接呼叫"""
        print(f"任務 {self.task_id} 開始處理 {self.workload} 項工作")
        total = 0
        for i in range(self.workload):
            total += i * i
            time.sleep(0.01)
        self.result = total
        print(f"任務 {self.task_id} 完成，結果: {self.result}")

# 使用
if __name__ == "__main__":
    tasks = [TaskProcess(i, 100 + i * 50) for i in range(4)]
    for t in tasks: t.start()
    for t in tasks: t.join()
    
    for t in tasks:
        print(f"任務 {t.task_id} 回傳: {t.result}")

進階：進程池 Pool - 管理工作者群


from multiprocessing import Pool
import time

def cpu_intensive_task(n):
    """CPU 密集型任務範例"""
    total = 0
    for i in range(n * 10000):
        total += i ** 2
    return total

if __name__ == "__main__":
    # 建立進程池 (預設 CPU 核心數)
    with Pool(processes=4) as pool:
        # 同步映射：阻塞直到全部完成
        results = pool.map(cpu_intensive_task, [10, 20, 30, 40, 50])
        print("map 結果:", results)
        
        # 非同步映射：立即回傳 AsyncResult
        async_result = pool.map_async(cpu_intensive_task, [5, 15, 25])
        print("非同步提交完成，繼續其他工作...")
        results = async_result.get(timeout=30)  # 等待結果
        print("map_async 結果:", results)
        
        # 非同步單一任務
        futures = [pool.apply_async(cpu_intensive_task, (i,)) for i in range(5, 10)]
        results = [f.get() for f in futures]
        print("apply_async 結果:", results)
        
        # imap: 惰性迭代器，記憶體效率高
        for result in pool.imap(cpu_intensive_task, range(3, 8)):
            print(f"imap 產出: {result}")
        
        # imap_unordered: 完成順序產出，更快
        for result in pool.imap_unordered(cpu_intensive_task, range(3, 8)):
            print(f"imap_unordered 產出: {result}")

共享狀態：Value、Array、Manager


from multiprocessing import Process, Value, Array, Manager, Lock
import ctypes

# 1. 簡單共享值 (原子操作需鎖)
def increment_shared(counter, lock, times):
    for _ in range(times):
        with lock:
            counter.value += 1

# 2. 共享陣列
def modify_shared_array(arr, idx, val):
    arr[idx] = val

# 3. Manager: 支援複雜物件 (dict, list, 自定義類別)
def manager_worker(shared_dict, shared_list, key):
    shared_dict[key] = f"value_{key}"
    shared_list.append(key)

if __name__ == "__main__":
    # Value + Lock
    counter = Value('i', 0)  # 'i' = signed int
    lock = Lock()
    processes = [Process(target=increment_shared, args=(counter, lock, 10000)) for _ in range(4)]
    for p in processes: p.start()
    for p in processes: p.join()
    print(f"計數器結果: {counter.value}")  # 應為 40000
    
    # Array
    arr = Array('d', [0.0] * 10)  # 'd' = double
    p = Process(target=modify_shared_array, args=(arr, 5, 3.14))
    p.start(); p.join()
    print(f"陣列[5] = {arr[5]}")
    
    # Manager (進程間共享複雜物件，透過 Proxy 代理)
    with Manager() as manager:
        shared_dict = manager.dict()
        shared_list = manager.list()
        
        procs = [Process(target=manager_worker, args=(shared_dict, shared_list, i)) for i in range(5)]
        for p in procs: p.start()
        for p in procs: p.join()
        
        print(f"共享字典: {dict(shared_dict)}")
        print(f"共享列表: {list(shared_list)}")

共享記憶體 (Python 3.8+)：高效能數值計算


from multiprocessing import shared_memory
import numpy as np
import os

def worker_shm(shm_name, shape, dtype, worker_id):
    # 連接現有共享記憶體
    existing_shm = shared_memory.SharedMemory(name=shm_name)
    arr = np.ndarray(shape, dtype=dtype, buffer=existing_shm.buf)
    
    # 執行計算 (例如: 每個進程處理一部分)
    chunk_size = shape[0] // 4
    start = worker_id * chunk_size
    end = start + chunk_size if worker_id < 3 else shape[0]
    
    arr[start:end] = arr[start:end] * 2  # 就地修改
    
    existing_shm.close()  # 釋放連接 (不銷毀記憶體)

if __name__ == "__main__":
    # 建立共享記憶體
    data = np.random.rand(1000000).astype(np.float64)
    shm = shared_memory.SharedMemory(create=True, size=data.nbytes)
    shared_arr = np.ndarray(data.shape, dtype=data.dtype, buffer=shm.buf)
    shared_arr[:] = data[:]  # 複製資料
    
    print(f"原始資料前5: {shared_arr[:5]}")
    
    processes = []
    for i in range(4):
        p = Process(target=worker_shm, args=(shm.name, data.shape, data.dtype, i))
        processes.append(p)
        p.start()
    
    for p in processes: p.join()
    
    print(f"處理後前5: {shared_arr[:5]}")
    
    # 清理
    shm.close()
    shm.unlink()  # 實際銷毀共享記憶體

進程池進階模式：動態調整、優雅關閉、錯誤處理


from multiprocessing import Pool, cpu_count
import signal
import sys
import atexit

class GracefulPool:
    """支援優雅關閉的進程池封裝"""
    
    def __init__(self, processes=None, initializer=None, initargs=()):
        self._pool = Pool(processes or cpu_count(), initializer, initargs)
        self._shutdown = False
        # 註冊信號處理
        signal.signal(signal.SIGINT, self._signal_handler)
        signal.signal(signal.SIGTERM, self._signal_handler)
        atexit.register(self.close)
    
    def _signal_handler(self, signum, frame):
        print(f"收到信號 {signum}，開始優雅關閉...")
        self.close()
        sys.exit(0)
    
    def map(self, func, iterable, chunksize=1):
        if self._shutdown:
            raise RuntimeError("進程池已關閉")
        return self._pool.map(func, iterable, chunksize)
    
    def map_async(self, func, iterable, chunksize=1, callback=None):
        if self._shutdown:
            raise RuntimeError("進程池已關閉")
        return self._pool.map_async(func, iterable, chunksize, callback)
    
    def apply_async(self, func, args=(), kwds={}, callback=None):
        if self._shutdown:
            raise RuntimeError("進程池已關閉")
        return self._pool.apply_async(func, args, kwds, callback)
    
    def close(self):
        if not self._shutdown:
            self._shutdown = True
            self._pool.close()
            self._pool.join()
    
    def terminate(self):
        self._pool.terminate()
        self._pool.join()
    
    def __enter__(self):
        return self
    
    def __exit__(self, *args):
        self.close()

# 使用範例
def risky_task(x):
    if x == 3:
        raise ValueError("故意錯誤")
    return x * x

with GracefulPool(4) as pool:
    try:
        results = pool.map(risky_task, range(10))
    except Exception as e:
        print(f"捕獲錯誤: {e}")
    # 上下文管理器自動 close/join

Process vs Pool vs Threading 比較

特性	multiprocessing.Process	multiprocessing.Pool	threading.Thread
並行類型	真並行 (多核心)	真並行 (多核心)	偽並行 (GIL 限制)
記憶體隔離	完全隔離	完全隔離	共享記憶體
啟動開銷	高 (~50-100ms)	高 (池化複用)	低 (~1ms)
適合任務	長時間、獨立、少量	大量同質、CPU 密集	I/O 密集、高併發
資料共享	需顯式 IPC	需顯式 IPC	直接共享 (需鎖)
錯誤隔離	進程崩潰不影響主程式	工作者崩潰自動重啟	執行緒崩潰可能拖垮進程

最佳實務檢查清單

✅ 所有進程建立代碼在 `if __name__ == '__main__':` 保護下
✅ 使用 `with Pool() as pool:` 上下文管理器確保資源釋放
✅ CPU 密集型任務進程數設為 `cpu_count()`，I/O 密集可設 2-4x
✅ 大量任務使用 `chunksize` 減少 IPC 開銷 (建議 `len(iterable) / (4 * processes)`)
✅ 避免在進程間傳遞大物件，使用共享記憶體或檔案路徑
✅ 異常處理：`apply_async` 需在 `get()` 時捕獲，`map` 直接拋出
✅ 設定 `maxtasksperchild` 定期回收工作者進程 (防記憶體洩漏)
✅ 使用 `initializer`/`initargs` 初始化工作者狀態 (資料庫連線、設定)

常見效能陷阱

陷阱	症狀	解決
進程數過多	CPU 切換開銷大、記憶體不足	限制為 CPU 核心數
頻繁建立/銷毀進程	效能極差	使用 Pool 複用
傳遞大量資料給工作者	Pickle 序列化成為瓶頸	共享記憶體、檔案、資料庫
任務粒度過細	IPC 開銷超過計算時間	批次處理、增加 chunksize
共享鎖競爭激烈	並行度無法提升	無鎖資料結構、分區處理、ThreadLocal