Python使用multiprocessing模塊實現(xiàn)多進(jìn)程并行計算
更新時間：2025年07月21日 09:44:51 作者：彬彬俠
Python的multiprocessing模塊是一個標(biāo)準(zhǔn)庫模塊,用于實現(xiàn)多進(jìn)程并行計算,相比線程（threading 模塊）,multiprocessing更適合需要高性能計算的場景,本文將詳細(xì)介紹multiprocessing模塊的定義、功能、用法、示例、應(yīng)用場景、最佳實踐和注意事項,需要的朋友可以參考下
引言
Python 的 multiprocessing 模塊是一個標(biāo)準(zhǔn)庫模塊，用于實現(xiàn)多進(jìn)程并行計算。它通過創(chuàng)建獨立的進(jìn)程，繞過 Python 的全局解釋器鎖（GIL），在多核 CPU 上實現(xiàn)真正的并行，特別適合 CPU 密集型任務(wù)（如數(shù)值計算、圖像處理）。相比線程（threading 模塊），multiprocessing 更適合需要高性能計算的場景。本文將詳細(xì)介紹 multiprocessing 模塊的定義、功能、用法、示例、應(yīng)用場景、最佳實踐和注意事項。
1. multiprocessing 模塊的定義和原理
1.1 定義
multiprocessing 是一個跨平臺的模塊，提供創(chuàng)建和管理進(jìn)程的 API，支持進(jìn)程間通信（IPC）、同步機(jī)制和共享資源管理。它模仿了 threading 模塊的接口，方便開發(fā)者從線程遷移到進(jìn)程。
核心功能：
- 進(jìn)程創(chuàng)建：創(chuàng)建獨立進(jìn)程，運行指定函數(shù)或任務(wù)。
- 進(jìn)程池：管理一組工作進(jìn)程，分配任務(wù)。
- 進(jìn)程通信：支持管道（
Pipe）、隊列（Queue）等 IPC 機(jī)制。 - 同步原語：提供鎖（
Lock）、信號量（Semaphore）、事件（Event）等。 - 共享內(nèi)存：支持共享基本數(shù)據(jù)類型（
Value）和數(shù)組（Array）。 - 跨平臺：在 Windows、Linux、macOS 上運行一致。
依賴：標(biāo)準(zhǔn)庫，無需額外安裝。
1.2 原理
- 進(jìn)程 vs 線程：
- 進(jìn)程：獨立的內(nèi)存空間，擁有自己的 Python 解釋器和 GIL，適合 CPU 密集型任務(wù)。
- 線程：共享內(nèi)存空間，受 GIL 限制，適合 I/O 密集型任務(wù)。
- GIL 繞過：每個進(jìn)程有獨立的 GIL，允許多核并行。
- 進(jìn)程創(chuàng)建：
- Linux/macOS：使用
fork（復(fù)制父進(jìn)程），或spawn（新進(jìn)程）。 - Windows：始終使用
spawn，啟動新解釋器。
- Linux/macOS：使用
- 通信開銷：進(jìn)程間通信（如
Queue）比線程慢，需優(yōu)化設(shè)計。
1.3 導(dǎo)入
import multiprocessing
2. multiprocessing 的核心組件和功能
2.1 進(jìn)程創(chuàng)建（Process）
通過 multiprocessing.Process 創(chuàng)建進(jìn)程，運行指定函數(shù)。
構(gòu)造函數(shù)：
Process(target=None, args=(), kwargs={}, name=None, daemon=None)
target：目標(biāo)函數(shù)。args/kwargs：函數(shù)參數(shù)。name：進(jìn)程名稱。daemon：是否為守護(hù)進(jìn)程（隨主進(jìn)程退出）。
主要方法：
start()：啟動進(jìn)程。join()：等待進(jìn)程結(jié)束。terminate()：強制終止進(jìn)程。is_alive()：檢查進(jìn)程是否存活。
示例：
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 個進(jìn)程，每個運行
worker函數(shù)。
2.2 進(jìn)程池（Pool）
Pool 用于管理固定數(shù)量的進(jìn)程，適合并行處理大量任務(wù)。
構(gòu)造函數(shù)：
Pool(processes=None, initializer=None, initargs=())
processes：進(jìn)程數(shù)（默認(rèn) CPU 核心數(shù)）。initializer：每個進(jìn)程的初始化函數(shù)。initargs：初始化函數(shù)參數(shù)。
主要方法：
map(func, iterable)：并行執(zhí)行func應(yīng)用于iterable，返回結(jié)果列表。imap(func, iterable)：惰性版本，返回迭代器。apply(func, args=(), kwds={})：同步執(zhí)行單任務(wù)。apply_async(func, args=(), kwds={})：異步執(zhí)行單任務(wù)。close()：關(guān)閉池，禁止新任務(wù)。join()：等待池內(nèi)進(jìn)程完成。
示例：
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 進(jìn)程通信
支持 Pipe 和 Queue 實現(xiàn)進(jìn)程間數(shù)據(jù)交換。
Pipe
- 雙向或單向管道，適合兩個進(jìn)程通信。
構(gòu)造函數(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
- 線程和進(jìn)程安全的隊列，適合多生產(chǎn)者/消費者場景。
構(gòu)造函數(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 同步機(jī)制
提供鎖、信號量等原語，確保進(jìn)程安全訪問共享資源。
Lock
- 互斥鎖，防止多個進(jìn)程同時訪問資源。
- 示例：
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 個進(jìn)程同時訪問
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
- 進(jìn)程間信號通知。
- 示例：
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 共享內(nèi)存
通過 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. 應(yīng)用場景
數(shù)值計算：
- 并行處理矩陣運算、蒙特卡洛模擬。
- 示例：計算大數(shù)組的平方。
圖像處理：
- 并行處理圖像濾波、特征提取。
- 示例：批量應(yīng)用卷積濾波。
機(jī)器學(xué)習(xí)：
- 并行訓(xùn)練模型或處理數(shù)據(jù)預(yù)處理。
- 示例：并行特征提取。
數(shù)據(jù)處理：
- 并行處理 CSV 文件、數(shù)據(jù)庫查詢。
- 示例：多進(jìn)程解析日志文件。
爬蟲：
- 并行抓取網(wǎng)頁（注意網(wǎng)絡(luò)限制）。
- 示例：結(jié)合
urllib并發(fā)下載。
4. 示例：多進(jìn)程爬蟲
結(jié)合 urllib 和 Queue 實現(xiàn)并行網(wǎng)頁抓取。
示例：
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)重復(fù)導(dǎo)入模塊。
示例：
if __name__ == "__main__":
p = Process(target=worker)
p.start()
選擇進(jìn)程池：
- 對于批量任務(wù)，使用
Pool簡化管理。
示例：
with Pool(4) as pool:
results = pool.map(func, data)
優(yōu)化通信：
- 盡量減少進(jìn)程間通信，使用共享內(nèi)存或批量傳遞數(shù)據(jù)。
示例：
arr = Array("i", [0] * size)
異常處理：
- 在子進(jìn)程中捕獲異常，通過
Queue或日志返回。
示例：
def worker(queue):
try:
# 工作代碼
except Exception as e:
queue.put(str(e))
測試代碼：
- 使用
pytest測試多進(jìn)程行為。
示例：
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
進(jìn)程數(shù)選擇：
- 默認(rèn)使用 CPU 核心數(shù)（
multiprocessing.cpu_count()）。
示例：
processes = min(len(tasks), multiprocessing.cpu_count())
6. 注意事項
GIL 限制：
multiprocessing繞過 GIL，適合 CPU 密集型任務(wù)；I/O 密集型任務(wù)考慮threading或asyncio。
示例：
# I/O 密集型：使用 asyncio
import asyncio
async def fetch():
pass
Windows 兼容性：
- Windows 使用
spawn，需確保代碼在if __name__ == "__main__":中。
示例：
if __name__ == "__main__":
main()
資源管理：
- 及時關(guān)閉進(jìn)程和池，釋放資源。
示例：
with Pool() as pool:
pool.map(func, data)
序列化開銷：
- 傳遞大數(shù)據(jù)到子進(jìn)程（如通過
Queue）可能慢，使用共享內(nèi)存。
示例：
shared_data = Value("d", 0.0)
調(diào)試難度：
- 子進(jìn)程錯誤可能不易捕獲，使用日志或
Queue返回錯誤。
示例：
import logging logging.basicConfig(level=logging.INFO)
7. 總結(jié)
Python 的 multiprocessing 模塊是實現(xiàn)多進(jìn)程并行的強大工具，繞過 GIL，適合 CPU 密集型任務(wù)。其核心特點包括：
- 定義：提供進(jìn)程創(chuàng)建、通信、同步和共享內(nèi)存的 API。
- 功能：支持
Process、Pool、Queue、Pipe、Lock等。 - 應(yīng)用：數(shù)值計算、圖像處理、機(jī)器學(xué)習(xí)、數(shù)據(jù)處理、爬蟲。
- 最佳實踐：使用
if __name__ == "__main__":、優(yōu)化通信、測試代碼。
以上就是Python使用multiprocessing模塊實現(xiàn)多進(jìn)程并行計算的詳細(xì)內(nèi)容，更多關(guān)于Python multiprocessing多進(jìn)程并行計算的資料請關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
這篇文章主要介紹了python二分法的實現(xiàn)代碼，大家可以參考使用2013-11-11
Python函數(shù)參數(shù)類型*、**的區(qū)別
這篇文章主要介紹了Python函數(shù)參數(shù)類型*、**的區(qū)別,本文用實例講解它們的區(qū)別,并講解了閉包的相關(guān)知識,需要的朋友可以參考下2015-04-04
base編碼是一種加密解密措施,目前常用的有base16、base32和base64,以base64為例,base64加密后共有64中字符,本文給大家分享python使用base加密解密的相關(guān)知識,感興趣的朋友跟隨小編一起看看吧2024-05-05
本文主要介紹了opencv銀行卡號識別的項目實踐,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2025-04-04
Python+OpenCV圖片局部區(qū)域像素值處理改進(jìn)版詳解
這篇文章主要為大家詳細(xì)介紹了Python+OpenCV圖片局部區(qū)域像素值處理的改進(jìn)版，具有一定的參考價值，感興趣的小伙伴們可以參考一下2019-01-01
pycharm激活碼2020最新分享適用pycharm2020最新版親測可用
這篇文章主要介紹了pycharm激活碼2020最新分享適用pycharm2020最新版親測可用,同時也支持Intellij IDEA激活碼,PHPStorm激活碼大家可以放心使用需要的朋友可以參考下2020-11-11
Python中文分詞庫jieba(結(jié)巴分詞)詳細(xì)使用介紹
這篇文章主要介紹了Python中文分詞庫jieba(結(jié)巴分詞)提取詞,加載詞,修改詞頻,定義詞庫詳細(xì)使用介紹,需要的朋友可以參考下2022-04-04
今天小編就為大家分享一篇Pytorch之Variable的用法，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2019-12-12
Pandas的AB BA類型數(shù)據(jù)框去重復(fù)
這篇文章主要為大家介紹了Pandas的AB BA類型數(shù)據(jù)框去重復(fù)實現(xiàn)示例詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2023-05-05
這篇文章主要介紹了python 解決微分方程的操作(數(shù)值解法)，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-05-05