Python使用multiprocessing模塊實現(xiàn)多進程并行計算

更新時間：2025年07月21日 09:44:51 作者：彬彬俠

Python的multiprocessing模塊是一個標準庫模塊,用于實現(xiàn)多進程并行計算,相比線程（threading 模塊）,multiprocessing更適合需要高性能計算的場景,本文將詳細介紹multiprocessing模塊的定義、功能、用法、示例、應用場景、最佳實踐和注意事項,需要的朋友可以參考下

引言

Python 的 multiprocessing 模塊是一個標準庫模塊，用于實現(xiàn)多進程并行計算。它通過創(chuàng)建獨立的進程，繞過 Python 的全局解釋器鎖（GIL），在多核 CPU 上實現(xiàn)真正的并行，特別適合 CPU 密集型任務（如數(shù)值計算、圖像處理）。相比線程（threading 模塊），multiprocessing 更適合需要高性能計算的場景。本文將詳細介紹 multiprocessing 模塊的定義、功能、用法、示例、應用場景、最佳實踐和注意事項。

1. multiprocessing 模塊的定義和原理

1.1 定義

multiprocessing 是一個跨平臺的模塊，提供創(chuàng)建和管理進程的 API，支持進程間通信（IPC）、同步機制和共享資源管理。它模仿了 threading 模塊的接口，方便開發(fā)者從線程遷移到進程。

核心功能：

進程創(chuàng)建：創(chuàng)建獨立進程，運行指定函數(shù)或任務。
進程池：管理一組工作進程，分配任務。
進程通信：支持管道（Pipe）、隊列（Queue）等 IPC 機制。
同步原語：提供鎖（Lock）、信號量（Semaphore）、事件（Event）等。
共享內存：支持共享基本數(shù)據(jù)類型（Value）和數(shù)組（Array）。
跨平臺：在 Windows、Linux、macOS 上運行一致。

依賴：標準庫，無需額外安裝。

1.2 原理

進程 vs 線程：
- 進程：獨立的內存空間，擁有自己的 Python 解釋器和 GIL，適合 CPU 密集型任務。
- 線程：共享內存空間，受 GIL 限制，適合 I/O 密集型任務。
GIL 繞過：每個進程有獨立的 GIL，允許多核并行。
進程創(chuàng)建：
- Linux/macOS：使用 fork（復制父進程），或 spawn（新進程）。
- Windows：始終使用 spawn，啟動新解釋器。
通信開銷：進程間通信（如 Queue）比線程慢，需優(yōu)化設計。

1.3 導入

import multiprocessing

2. multiprocessing 的核心組件和功能

2.1 進程創(chuàng)建（Process）

通過 multiprocessing.Process 創(chuàng)建進程，運行指定函數(shù)。

構造函數(shù)：

Process(target=None, args=(), kwargs={}, name=None, daemon=None)

target：目標函數(shù)。
args/kwargs：函數(shù)參數(shù)。
name：進程名稱。
daemon：是否為守護進程（隨主進程退出）。

主要方法：

start()：啟動進程。
join()：等待進程結束。
terminate()：強制終止進程。
is_alive()：檢查進程是否存活。

示例：

import multiprocessing

def worker(num):
    print(f"Worker {num} running in process {multiprocessing.current_process().name}")

if __name__ == "__main__":
    processes = [multiprocessing.Process(target=worker, args=(i,)) for i in range(3)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()

輸出（順序可能不同）：

Worker 0 running in process Process-1
Worker 1 running in process Process-2
Worker 2 running in process Process-3

說明：創(chuàng)建 3 個進程，每個運行 worker 函數(shù)。

2.2 進程池（Pool）

Pool 用于管理固定數(shù)量的進程，適合并行處理大量任務。

構造函數(shù)：

Pool(processes=None, initializer=None, initargs=())

processes：進程數(shù)（默認 CPU 核心數(shù)）。
initializer：每個進程的初始化函數(shù)。
initargs：初始化函數(shù)參數(shù)。

主要方法：

map(func, iterable)：并行執(zhí)行 func 應用于 iterable，返回結果列表。
imap(func, iterable)：惰性版本，返回迭代器。
apply(func, args=(), kwds={})：同步執(zhí)行單任務。
apply_async(func, args=(), kwds={})：異步執(zhí)行單任務。
close()：關閉池，禁止新任務。
join()：等待池內進程完成。

示例：

from multiprocessing import Pool

def square(n):
    return n * n

if __name__ == "__main__":
    with Pool(processes=4) as pool:
        results = pool.map(square, range(10))
    print(results)  # 輸出: [0, 1, 4, 9, 16, 25, 36, 49, 64, 81]

2.3 進程通信

支持 Pipe 和 Queue 實現(xiàn)進程間數(shù)據(jù)交換。

Pipe

雙向或單向管道，適合兩個進程通信。

構造函數(shù)：

Pipe(duplex=True)

返回 (conn1, conn2)，兩個連接對象。
duplex=True：雙向；False：單向。

示例：

from multiprocessing import Process, Pipe

def sender(conn):
    conn.send("Hello from sender")
    conn.close()

def receiver(conn):
    print(conn.recv())
    conn.close()

if __name__ == "__main__":
    parent_conn, child_conn = Pipe()
    p1 = Process(target=sender, args=(child_conn,))
    p2 = Process(target=receiver, args=(parent_conn,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

輸出：

Hello from sender

Queue

線程和進程安全的隊列，適合多生產者/消費者場景。

構造函數(shù)：

Queue(maxsize=0)

maxsize：最大容量（0 表示無限制）。

示例：

from multiprocessing import Process, Queue

def producer(queue):
    queue.put("Data from producer")

def consumer(queue):
    print(queue.get())

if __name__ == "__main__":
    queue = Queue()
    p1 = Process(target=producer, args=(queue,))
    p2 = Process(target=consumer, args=(queue,))
    p1.start()
    p2.start()
    p1.join()
    p2.join()

2.4 同步機制

提供鎖、信號量等原語，確保進程安全訪問共享資源。

Lock

互斥鎖，防止多個進程同時訪問資源。
示例：

from multiprocessing import Process, Lock

def printer(lock, msg):
    with lock:
        print(msg)

if __name__ == "__main__":
    lock = Lock()
    processes = [Process(target=printer, args=(lock, f"Message {i}")) for i in range(3)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()

Semaphore

控制有限資源的并發(fā)訪問。
示例：

from multiprocessing import Process, Semaphore

def worker(sem, name):
    with sem:
        print(f"{name} acquired resource")
        # 模擬工作

if __name__ == "__main__":
    sem = Semaphore(2)  # 允許 2 個進程同時訪問
    processes = [Process(target=worker, args=(sem, f"Worker {i}")) for i in range(5)]
    for p in processes:
        p.start()
    for p in processes:
        p.join()

Event

進程間信號通知。
示例：

from multiprocessing import Process, Event
import time

def wait_for_event(event):
    event.wait()
    print("Event triggered")

if __name__ == "__main__":
    event = Event()
    p = Process(target=wait_for_event, args=(event,))
    p.start()
    time.sleep(1)
    event.set()  # 觸發(fā)事件
    p.join()

2.5 共享內存

通過 Value 和 Array 共享基本數(shù)據(jù)類型。

Value：單個共享值。
Array：共享數(shù)組。

示例：

from multiprocessing import Process, Value, Array

def modify(shared_num, shared_arr):
    shared_num.value += 1
    for i in range(len(shared_arr)):
        shared_arr[i] += 1

if __name__ == "__main__":
    num = Value("i", 0)  # 共享整數(shù)
    arr = Array("i", [1, 2, 3])  # 共享數(shù)組
    p = Process(target=modify, args=(num, arr))
    p.start()
    p.join()
    print(num.value)  # 輸出: 1
    print(list(arr))  # 輸出: [2, 3, 4]

3. 應用場景

數(shù)值計算：

并行處理矩陣運算、蒙特卡洛模擬。
示例：計算大數(shù)組的平方。

圖像處理：

并行處理圖像濾波、特征提取。
示例：批量應用卷積濾波。

機器學習：

并行訓練模型或處理數(shù)據(jù)預處理。
示例：并行特征提取。

數(shù)據(jù)處理：

并行處理 CSV 文件、數(shù)據(jù)庫查詢。
示例：多進程解析日志文件。

爬蟲：

并行抓取網頁（注意網絡限制）。
示例：結合 urllib 并發(fā)下載。

4. 示例：多進程爬蟲

結合 urllib 和 Queue 實現(xiàn)并行網頁抓取。

示例：

import urllib.request
from multiprocessing import Process, Queue
from urllib.error import URLError

def fetch_url(queue, url):
    try:
        with urllib.request.urlopen(url) as response:
            content = response.read().decode("utf-8")
            queue.put((url, len(content)))
    except URLError as e:
        queue.put((url, str(e)))

def main():
    urls = ["https://example.com", "https://python.org", "https://invalid-url"]
    queue = Queue()
    processes = [Process(target=fetch_url, args=(queue, url)) for url in urls]
    for p in processes:
        p.start()
    for p in processes:
        p.join()
    while not queue.empty():
        url, result = queue.get()
        print(f"{url}: {result}")

if __name__ == "__main__":
    main()

輸出（示例）：

https://example.com: 1256
https://python.org: 50000
https://invalid-url: [Errno 11001] getaddrinfo failed

5. 最佳實踐

使用 if __name__ == "__main__":：

防止 Windows 和某些 Unix 系統(tǒng)重復導入模塊。

示例：

if __name__ == "__main__":
    p = Process(target=worker)
    p.start()

選擇進程池：

對于批量任務，使用 Pool 簡化管理。

示例：

with Pool(4) as pool:
    results = pool.map(func, data)

優(yōu)化通信：

盡量減少進程間通信，使用共享內存或批量傳遞數(shù)據(jù)。

示例：

arr = Array("i", [0] * size)

異常處理：

在子進程中捕獲異常，通過 Queue 或日志返回。

示例：

def worker(queue):
    try:
        # 工作代碼
    except Exception as e:
        queue.put(str(e))

測試代碼：

使用 pytest 測試多進程行為。

示例：

import pytest
from multiprocessing import Process

def test_process():
    def worker():
        print("Test")
    p = Process(target=worker)
    p.start()
    p.join()
    assert p.exitcode == 0

進程數(shù)選擇：

默認使用 CPU 核心數(shù)（multiprocessing.cpu_count()）。

示例：

processes = min(len(tasks), multiprocessing.cpu_count())

6. 注意事項

GIL 限制：

multiprocessing 繞過 GIL，適合 CPU 密集型任務；I/O 密集型任務考慮 threading 或 asyncio。

示例：

# I/O 密集型：使用 asyncio
import asyncio
async def fetch():
    pass

Windows 兼容性：

Windows 使用 spawn，需確保代碼在 if __name__ == "__main__": 中。

示例：

if __name__ == "__main__":
    main()

資源管理：

及時關閉進程和池，釋放資源。

示例：

with Pool() as pool:
    pool.map(func, data)

序列化開銷：

傳遞大數(shù)據(jù)到子進程（如通過 Queue）可能慢，使用共享內存。

示例：

shared_data = Value("d", 0.0)

調試難度：

子進程錯誤可能不易捕獲，使用日志或 Queue 返回錯誤。

示例：

import logging
logging.basicConfig(level=logging.INFO)

7. 總結

Python 的 multiprocessing 模塊是實現(xiàn)多進程并行的強大工具，繞過 GIL，適合 CPU 密集型任務。其核心特點包括：

定義：提供進程創(chuàng)建、通信、同步和共享內存的 API。
功能：支持 Process、Pool、Queue、Pipe、Lock 等。
應用：數(shù)值計算、圖像處理、機器學習、數(shù)據(jù)處理、爬蟲。
最佳實踐：使用 if __name__ == "__main__":、優(yōu)化通信、測試代碼。

以上就是Python使用multiprocessing模塊實現(xiàn)多進程并行計算的詳細內容，更多關于Python multiprocessing多進程并行計算的資料請關注腳本之家其它相關文章！

您可能感興趣的文章:

python二分法實現(xiàn)實例
這篇文章主要介紹了python二分法的實現(xiàn)代碼，大家可以參考使用
2013-11-11
Python函數(shù)參數(shù)類型*、**的區(qū)別
這篇文章主要介紹了Python函數(shù)參數(shù)類型*、**的區(qū)別,本文用實例講解它們的區(qū)別,并講解了閉包的相關知識,需要的朋友可以參考下
2015-04-04
python如何使用base加密解密
base編碼是一種加密解密措施,目前常用的有base16、base32和base64,以base64為例,base64加密后共有64中字符,本文給大家分享python使用base加密解密的相關知識,感興趣的朋友跟隨小編一起看看吧
2024-05-05
opencv銀行卡號識別的項目實踐
本文主要介紹了opencv銀行卡號識別的項目實踐,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧
2025-04-04
Python+OpenCV圖片局部區(qū)域像素值處理改進版詳解
這篇文章主要為大家詳細介紹了Python+OpenCV圖片局部區(qū)域像素值處理的改進版，具有一定的參考價值，感興趣的小伙伴們可以參考一下
2019-01-01
pycharm激活碼2020最新分享適用pycharm2020最新版親測可用
這篇文章主要介紹了pycharm激活碼2020最新分享適用pycharm2020最新版親測可用,同時也支持Intellij IDEA激活碼,PHPStorm激活碼大家可以放心使用需要的朋友可以參考下
2020-11-11
Python中文分詞庫jieba(結巴分詞)詳細使用介紹
這篇文章主要介紹了Python中文分詞庫jieba(結巴分詞)提取詞,加載詞,修改詞頻,定義詞庫詳細使用介紹,需要的朋友可以參考下
2022-04-04
Pytorch之Variable的用法
今天小編就為大家分享一篇Pytorch之Variable的用法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧
2019-12-12
Pandas的AB BA類型數(shù)據(jù)框去重復
這篇文章主要為大家介紹了Pandas的AB BA類型數(shù)據(jù)框去重復實現(xiàn)示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪
2023-05-05
python 解決微分方程的操作(數(shù)值解法)
這篇文章主要介紹了python 解決微分方程的操作(數(shù)值解法)，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教
2021-05-05