除了核心的multiprocessing模塊，Python還有多款適用于不同場景的多進(jìn)程實(shí)現(xiàn)工具——有的輕量化、有的適配分布式、有的兼容Shell命令，本文結(jié)合實(shí)際業(yè)務(wù)場景詳解這些模塊的用法、優(yōu)勢和適用場景，幫你根據(jù)需求選對(duì)工具。

一、subprocess：調(diào)用外部進(jìn)程（輕量首選）

核心定位

subprocess是Python內(nèi)置模塊，核心作用是啟動(dòng)外部進(jìn)程（如執(zhí)行Shell命令、調(diào)用其他可執(zhí)行文件），而非像multiprocessing那樣在Python代碼內(nèi)創(chuàng)建子進(jìn)程執(zhí)行函數(shù)，是“跨語言/跨程序”多進(jìn)程的輕量方案。

核心優(yōu)勢

• 內(nèi)置模塊無需安裝，語法簡單；
• 可直接調(diào)用系統(tǒng)命令（如ls、ping）或其他語言腳本（如Shell、C++可執(zhí)行文件）；
• 支持捕獲外部進(jìn)程的輸出、輸入和錯(cuò)誤信息。

實(shí)操場景：批量執(zhí)行Shell命令

import subprocess
import os

# 場景：并行執(zhí)行多個(gè)ping命令（檢測多個(gè)服務(wù)器連通性）
def ping_host(host):
    # 執(zhí)行ping命令，捕獲輸出
    result = subprocess.run(
        ["ping", "-c", "2", host],  # 命令參數(shù)列表（避免Shell注入）
        stdout=subprocess.PIPE,     # 捕獲標(biāo)準(zhǔn)輸出
        stderr=subprocess.PIPE,     # 捕獲標(biāo)準(zhǔn)錯(cuò)誤
        encoding="utf-8"            # 輸出編碼為字符串（無需手動(dòng)解碼）
    )
    # 解析結(jié)果
    if result.returncode == 0:
        return f"{host}：連通"
    else:
        return f"{host}：不通（{result.stderr[:50]}）"

if __name__ == "__main__":
    hosts = ["baidu.com", "github.com", "192.168.1.1"]
    # 并行執(zhí)行（通過多線程調(diào)用subprocess，底層是系統(tǒng)多進(jìn)程）
    from concurrent.futures import ThreadPoolExecutor
    with ThreadPoolExecutor(max_workers=3) as executor:
        results = executor.map(ping_host, hosts)
    for res in results:
        print(res)

執(zhí)行結(jié)果：

baidu.com：連通
github.com：連通
192.168.1.1：不通（ping: connect: Network is unreachable）

適用場景

• 調(diào)用系統(tǒng)命令、外部腳本/可執(zhí)行文件；
• 簡單的多進(jìn)程任務(wù)，無需復(fù)雜的進(jìn)程間通信；
• 需捕獲外部程序輸出/錯(cuò)誤的場景。

二、concurrent.futures.ProcessPoolExecutor：簡化進(jìn)程池（高層封裝）

核心定位

concurrent.futures.ProcessPoolExecutor是Python3.2+內(nèi)置的高層封裝，底層基于multiprocessing實(shí)現(xiàn)，但API更簡潔，無需手動(dòng)管理進(jìn)程池的close()/join()，是“開箱即用”的進(jìn)程池方案。

核心優(yōu)勢

• 語法極簡，一行代碼提交批量任務(wù)；
• 內(nèi)置異常處理，無需手動(dòng)捕獲子進(jìn)程錯(cuò)誤；
• 支持map()/submit()兩種提交方式，兼容迭代器參數(shù)。

實(shí)操場景：批量計(jì)算（替代multiprocessing.Pool）

from concurrent.futures import ProcessPoolExecutor
import time

# 耗時(shí)計(jì)算任務(wù)
def calculate_square(num):
    time.sleep(1)
    return num * num

if __name__ == "__main__":
    start_time = time.time()
    # 創(chuàng)建進(jìn)程池（自動(dòng)適配CPU核心數(shù)）
    with ProcessPoolExecutor() as executor:
        # 批量提交任務(wù)（range(10)為參數(shù)列表）
        results = executor.map(calculate_square, range(10))
    # 輸出結(jié)果
    print("計(jì)算結(jié)果：", list(results))
    print(f"總耗時(shí)：{time.time() - start_time:.2f}秒")

執(zhí)行結(jié)果：

計(jì)算結(jié)果： [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]
總耗時(shí)：3.01秒

關(guān)鍵對(duì)比（vs multiprocessing.Pool）

適用場景

• 簡單的CPU密集型批量任務(wù)；
• 追求代碼簡潔，無需復(fù)雜進(jìn)程通信；
• Python3.2+環(huán)境，希望減少冗余代碼。

三、os.fork()：Unix系原生多進(jìn)程（底層實(shí)現(xiàn)）

核心定位

os.fork()是Unix/Linux/Mac系統(tǒng)特有的底層調(diào)用，直接復(fù)制當(dāng)前進(jìn)程（父進(jìn)程→子進(jìn)程），是Python多進(jìn)程的“底層基礎(chǔ)”（multiprocessing底層也基于fork()）。

核心優(yōu)勢

• 極致輕量，無額外封裝開銷；
• 完全掌控進(jìn)程創(chuàng)建/銷毀流程；
• 適合深入理解多進(jìn)程原理。

實(shí)操場景：簡單父子進(jìn)程通信

import os
import time

if __name__ == "__main__":
    # 創(chuàng)建子進(jìn)程（僅Unix系可用）
    pid = os.fork()
    
    if pid == 0:
        # 子進(jìn)程邏輯
        print(f"子進(jìn)程：PID={os.getpid()}，父進(jìn)程PID={os.getppid()}")
        time.sleep(2)
        print("子進(jìn)程執(zhí)行完成")
        os._exit(0)  # 子進(jìn)程退出
    else:
        # 父進(jìn)程邏輯
        print(f"父進(jìn)程：創(chuàng)建子進(jìn)程PID={pid}")
        # 等待子進(jìn)程結(jié)束，獲取退出狀態(tài)
        wait_pid, status = os.waitpid(pid, 0)
        print(f"父進(jìn)程：子進(jìn)程{wait_pid}退出，狀態(tài)={status}")

執(zhí)行結(jié)果（Linux系統(tǒng)）：

父進(jìn)程：創(chuàng)建子進(jìn)程PID=12346
子進(jìn)程：PID=12346，父進(jìn)程PID=12345
子進(jìn)程執(zhí)行完成
父進(jìn)程：子進(jìn)程12346退出，狀態(tài)=0

注意事項(xiàng)

• ? Windows系統(tǒng)不支持os.fork()（無fork系統(tǒng)調(diào)用）；
• 子進(jìn)程會(huì)復(fù)制父進(jìn)程的所有內(nèi)存空間，需注意資源占用；
• 需手動(dòng)管理子進(jìn)程退出（os._exit()），避免僵尸進(jìn)程。

適用場景

• Unix/Linux環(huán)境下的底層進(jìn)程控制；
• 需極致性能，無Python封裝開銷；
• 學(xué)習(xí)多進(jìn)程底層原理。

四、celery：分布式多進(jìn)程（大規(guī)模任務(wù)）

核心定位

celery是分布式任務(wù)隊(duì)列框架，基于multiprocessing/gevent實(shí)現(xiàn)多進(jìn)程/多線程，支持跨機(jī)器、跨語言的分布式任務(wù)調(diào)度，是“大規(guī)模、分布式”多進(jìn)程的首選。

核心優(yōu)勢

• 支持分布式部署（多臺(tái)機(jī)器協(xié)同）；
• 內(nèi)置任務(wù)重試、定時(shí)任務(wù)、任務(wù)優(yōu)先級(jí)；
• 兼容Redis/RabbitMQ等消息中間件。

前置準(zhǔn)備

# 安裝celery和Redis（消息中間件）
pip install celery redis

實(shí)操場景：分布式任務(wù)執(zhí)行

1. 定義任務(wù)（tasks.py）

from celery import Celery

# 初始化Celery，使用Redis作為消息中間件
app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",  # 任務(wù)隊(duì)列
    backend="redis://localhost:6379/1"   # 結(jié)果存儲(chǔ)
)

# 定義耗時(shí)任務(wù)
@app.task
def calculate_sum(a, b):
    import time
    time.sleep(2)
    return a + b

2. 啟動(dòng)worker（多進(jìn)程執(zhí)行任務(wù)）

# 啟動(dòng)celery worker，指定4個(gè)進(jìn)程
celery -A tasks worker --loglevel=info --concurrency=4

3. 提交任務(wù)（main.py）

from tasks import calculate_sum

if __name__ == "__main__":
    # 異步提交多個(gè)任務(wù)
    task1 = calculate_sum.delay(10, 20)
    task2 = calculate_sum.delay(30, 40)
    
    # 獲取結(jié)果（阻塞等待）
    print("任務(wù)1結(jié)果：", task1.get())
    print("任務(wù)2結(jié)果：", task2.get())

執(zhí)行結(jié)果：

任務(wù)1結(jié)果：30
任務(wù)2結(jié)果：70

適用場景

• 分布式系統(tǒng)（多機(jī)器執(zhí)行任務(wù)）；
• 大規(guī)模異步任務(wù)（如定時(shí)任務(wù)、批量數(shù)據(jù)處理）；
• 需任務(wù)重試、優(yōu)先級(jí)、結(jié)果持久化的場景。

五、joblib：科學(xué)計(jì)算場景多進(jìn)程（適配numpy/pandas）

核心定位

joblib是專為科學(xué)計(jì)算設(shè)計(jì)的庫，優(yōu)化了大數(shù)組的多進(jìn)程傳輸（避免數(shù)據(jù)拷貝），兼容numpy/pandas/scikit-learn等庫，是數(shù)據(jù)科學(xué)領(lǐng)域的多進(jìn)程首選。

核心優(yōu)勢

• 高效傳輸大數(shù)組（共享內(nèi)存，無需拷貝）；
• 語法簡潔，適配科學(xué)計(jì)算場景；
• 支持緩存計(jì)算結(jié)果（避免重復(fù)計(jì)算）。

前置準(zhǔn)備

pip install joblib numpy

實(shí)操場景：批量處理numpy數(shù)組

from joblib import Parallel, delayed
import numpy as np
import time

# 處理大數(shù)組的任務(wù)
def process_array(arr):
    time.sleep(1)
    return np.mean(arr)  # 計(jì)算數(shù)組均值

if __name__ == "__main__":
    start_time = time.time()
    # 生成10個(gè)大數(shù)組（每個(gè)100萬元素）
    arrays = [np.random.rand(1000000) for _ in range(10)]
    # 多進(jìn)程處理（n_jobs=4為進(jìn)程數(shù)）
    results = Parallel(n_jobs=4)(delayed(process_array)(arr) for arr in arrays)
    # 輸出結(jié)果
    print("數(shù)組均值列表：", [round(r, 4) for r in results])
    print(f"總耗時(shí)：{time.time() - start_time:.2f}秒")

執(zhí)行結(jié)果：

數(shù)組均值列表： [0.5001, 0.4998, 0.5002, 0.4999, 0.5000, 0.5001, 0.4997, 0.5003, 0.4999, 0.5000]
總耗時(shí)：3.02秒

關(guān)鍵優(yōu)勢（vs multiprocessing）

• 處理numpy/pandas大數(shù)組時(shí)，joblib通過共享內(nèi)存?zhèn)鬏?，速度?code>multiprocessing快50%+；
• 內(nèi)置緩存機(jī)制（Memory類），可緩存重復(fù)計(jì)算的結(jié)果，適合迭代開發(fā)。

適用場景

• 數(shù)據(jù)科學(xué)/機(jī)器學(xué)習(xí)場景（numpy/pandas/scikit-learn）；
• 處理大數(shù)組、避免數(shù)據(jù)拷貝；
• 需緩存計(jì)算結(jié)果的迭代開發(fā)。

六、multiprocess：增強(qiáng)版multiprocessing（第三方）

核心定位

multiprocess是multiprocessing的第三方增強(qiáng)版，修復(fù)了原版的多個(gè)bug，新增了更多實(shí)用功能（如進(jìn)程池超時(shí)、更友好的API）。

核心優(yōu)勢

• 完全兼容multiprocessing的API，可直接替換；
• 新增進(jìn)程池超時(shí)、異步結(jié)果回調(diào)等功能；
• 修復(fù)了原版在Windows系統(tǒng)的部分兼容性問題。

前置準(zhǔn)備

pip install multiprocess

實(shí)操場景：帶超時(shí)的進(jìn)程池

from multiprocess import Pool
import time

def slow_task(num):
    time.sleep(5)  # 耗時(shí)5秒的任務(wù)
    return num * num

if __name__ == "__main__":
    with Pool(processes=2) as pool:
        # 提交任務(wù)，設(shè)置超時(shí)3秒
        try:
            result = pool.apply_async(slow_task, args=(10,)).get(timeout=3)
            print("任務(wù)結(jié)果：", result)
        except TimeoutError:
            print("任務(wù)超時(shí)，終止執(zhí)行")

執(zhí)行結(jié)果：

任務(wù)超時(shí)，終止執(zhí)行

適用場景

• 需兼容multiprocessing但需要更多功能；
• Windows系統(tǒng)下multiprocessing兼容性問題；
• 需進(jìn)程池超時(shí)、回調(diào)等增強(qiáng)功能。

七、各模塊選型指南（按場景匹配）

總結(jié)：核心選型原則

1. 基礎(chǔ)場景優(yōu)先：無特殊需求時(shí)，multiprocessing仍是首選（功能全、文檔完善）；
2. 簡潔優(yōu)先：簡單批量任務(wù)用concurrent.futures.ProcessPoolExecutor，減少冗余代碼；
3. 場景適配：數(shù)據(jù)科學(xué)選joblib，分布式選celery，調(diào)用外部程序選subprocess；
4. 平臺(tái)兼容：Windows系統(tǒng)避免os.fork()，優(yōu)先用multiprocessing/concurrent.futures。