Python 如何使用 multiprocessing 模塊創(chuàng)建進(jìn)程池

一、簡(jiǎn)介

在現(xiàn)代計(jì)算中，提升程序性能的一個(gè)關(guān)鍵方法是并行處理，尤其是當(dāng)處理大量數(shù)據(jù)或計(jì)算密集型任務(wù)時(shí)，單線程可能不夠高效。Python 提供了多個(gè)模塊來(lái)支持并行計(jì)算，其中最常用的就是 multiprocessing 模塊。它允許我們?cè)诙鄠€(gè)處理器上同時(shí)運(yùn)行代碼，通過多個(gè)進(jìn)程同時(shí)處理任務(wù)，極大地提高了效率。

本文將介紹如何使用 Python 中的 multiprocessing 模塊，特別是 進(jìn)程池 的概念。我們會(huì)講解如何創(chuàng)建進(jìn)程池并在其上分配任務(wù)，通過代碼示例幫助你輕松理解這一重要技術(shù)。

二、進(jìn)程池簡(jiǎn)介

2.1 什么是進(jìn)程池？

進(jìn)程池（Process Pool） 是指通過預(yù)先創(chuàng)建的一組進(jìn)程來(lái)并發(fā)執(zhí)行任務(wù)。通常情況下，系統(tǒng)的進(jìn)程創(chuàng)建和銷毀是非常耗時(shí)的，所以使用進(jìn)程池可以避免頻繁的創(chuàng)建和銷毀開銷。我們可以將任務(wù)提交給進(jìn)程池，讓它們分配給預(yù)先啟動(dòng)的進(jìn)程進(jìn)行處理。

進(jìn)程池最常用于：

• 大量任務(wù)需要并行執(zhí)行時(shí)。
• 避免頻繁的進(jìn)程創(chuàng)建和銷毀。
• 有限的系統(tǒng)資源，例如 CPU 核心數(shù)有限時(shí)，通過控制池的大小來(lái)限制并發(fā)進(jìn)程數(shù)。

2.2 為什么使用進(jìn)程池？

在 Python 中，由于 GIL（Global Interpreter Lock，全局解釋器鎖） 的存在，線程并發(fā)無(wú)法在 CPU 密集型任務(wù)中充分發(fā)揮多核優(yōu)勢(shì)。multiprocessing 模塊通過多進(jìn)程方式繞過 GIL 限制，使得程序能夠充分利用多核 CPU 的優(yōu)勢(shì)。相比于手動(dòng)創(chuàng)建和管理多個(gè)進(jìn)程，使用進(jìn)程池能讓我們更輕松地管理并發(fā)任務(wù)。

進(jìn)程池的優(yōu)點(diǎn)包括：

• 自動(dòng)管理多個(gè)進(jìn)程的創(chuàng)建和銷毀。
• 可以方便地并行執(zhí)行多個(gè)任務(wù)。
• 通過池大小控制并發(fā)的進(jìn)程數(shù)量，避免資源過度占用。

三、使用 multiprocessing模塊的基礎(chǔ)知識(shí)

在開始使用進(jìn)程池之前，了解 Python 中 multiprocessing 模塊的基本概念很重要。

3.1 創(chuàng)建和啟動(dòng)進(jìn)程

在 multiprocessing 模塊中，我們可以通過 Process 類創(chuàng)建和啟動(dòng)進(jìn)程。簡(jiǎn)單示例如下：

import multiprocessing
import time
def worker(num):
    """ 工作函數(shù)，執(zhí)行一些任務(wù) """
    print(f"Worker {num} is starting")
    time.sleep(2)  # 模擬工作
    print(f"Worker {num} is done")
if __name__ == '__main__':
    processes = []
    for i in range(5):
        p = multiprocessing.Process(target=worker, args=(i,))
        processes.append(p)
        p.start()
    for p in processes:
        p.join()  # 等待所有進(jìn)程完成

這個(gè)示例演示了如何創(chuàng)建多個(gè)進(jìn)程并并行執(zhí)行任務(wù)，但當(dāng)任務(wù)數(shù)很多時(shí)，手動(dòng)管理這些進(jìn)程就顯得復(fù)雜了。這時(shí)，進(jìn)程池就派上了用場(chǎng)。

四、創(chuàng)建進(jìn)程池并分配任務(wù)

4.1 Pool 類的基本用法

multiprocessing 模塊中的 Pool 類提供了一種方便的方式來(lái)創(chuàng)建進(jìn)程池并分配任務(wù)。我們可以將多個(gè)任務(wù)提交給進(jìn)程池，由進(jìn)程池中的多個(gè)進(jìn)程同時(shí)處理。

以下是使用 Pool 創(chuàng)建進(jìn)程池并執(zhí)行任務(wù)的基本示例：

import multiprocessing
import time
def worker(num):
    """ 工作函數(shù)，執(zhí)行任務(wù) """
    print(f"Worker {num} is starting")
    time.sleep(2)
    print(f"Worker {num} is done")
    return num * 2  # 返回計(jì)算結(jié)果
if __name__ == '__main__':
    # 創(chuàng)建包含 4 個(gè)進(jìn)程的進(jìn)程池
    with multiprocessing.Pool(processes=4) as pool:
        results = pool.map(worker, range(10))
    print(f"Results: {results}")

4.2 Pool.map() 方法

在上述代碼中，我們使用了 Pool.map() 方法。它的作用類似于 Python 內(nèi)置的 map() 函數(shù)，能夠?qū)⒁粋€(gè)可迭代對(duì)象的每個(gè)元素傳遞給目標(biāo)函數(shù)，并將結(jié)果以列表形式返回。Pool.map() 會(huì)自動(dòng)將任務(wù)分配給進(jìn)程池中的多個(gè)進(jìn)程并行處理。

例如：

• range(10) 生成了 10 個(gè)任務(wù)，每個(gè)任務(wù)調(diào)用一次 worker 函數(shù)。
• 由于進(jìn)程池中有 4 個(gè)進(jìn)程，所以它會(huì)一次并行執(zhí)行 4 個(gè)任務(wù)，直到所有任務(wù)完成。

4.3 其他常用方法

除了 map() 方法，Pool 類還有其他一些常用的方法：

apply()：同步執(zhí)行一個(gè)函數(shù)，直到該函數(shù)執(zhí)行完畢后，才能繼續(xù)執(zhí)行主進(jìn)程的代碼。

result = pool.apply(worker, args=(5,))

apply_async()：異步執(zhí)行一個(gè)函數(shù)，主進(jìn)程不會(huì)等待該函數(shù)執(zhí)行完畢，可以繼續(xù)執(zhí)行其他代碼。適合用于并行處理單個(gè)任務(wù)。

result = pool.apply_async(worker, args=(5,))
result.get()  # 獲取返回值

starmap()：類似 map()，但它允許傳遞多個(gè)參數(shù)給目標(biāo)函數(shù)。

def worker(a, b):
    return a + b
results = pool.starmap(worker, [(1, 2), (3, 4), (5, 6)])

4.4 進(jìn)程池大小的設(shè)置

在創(chuàng)建進(jìn)程池時(shí)，我們可以通過 processes 參數(shù)來(lái)設(shè)置進(jìn)程池的大小。通常，進(jìn)程池大小與系統(tǒng)的 CPU 核心數(shù)有關(guān)。你可以通過 multiprocessing.cpu_count() 方法獲取當(dāng)前系統(tǒng)的 CPU 核心數(shù)，然后根據(jù)需要設(shè)置進(jìn)程池的大小。

import multiprocessing
# 獲取系統(tǒng) CPU 核心數(shù)
cpu_count = multiprocessing.cpu_count()
# 創(chuàng)建進(jìn)程池，進(jìn)程數(shù)與 CPU 核心數(shù)相同
pool = multiprocessing.Pool(processes=cpu_count)

將進(jìn)程池大小設(shè)置為與 CPU 核心數(shù)相同是一個(gè)常見的選擇，因?yàn)檫@可以充分利用系統(tǒng)資源。

五、進(jìn)程池的高級(jí)用法

5.1 異步任務(wù)處理

在實(shí)際場(chǎng)景中，某些任務(wù)可能會(huì)耗時(shí)較長(zhǎng)。如果我們不希望等待這些任務(wù)完成再執(zhí)行其他代碼，可以使用異步任務(wù)處理方法，如 apply_async()。它允許我們?cè)诤笈_(tái)執(zhí)行任務(wù)，而主進(jìn)程可以繼續(xù)執(zhí)行其他代碼，任務(wù)完成后我們可以通過 result.get() 獲取結(jié)果。

import multiprocessing
import time
def worker(num):
    time.sleep(2)
    return num * 2
if __name__ == '__main__':
    with multiprocessing.Pool(processes=4) as pool:
        results = [pool.apply_async(worker, args=(i,)) for i in range(10)]
        # 執(zhí)行其他操作
        print("主進(jìn)程繼續(xù)運(yùn)行")
        # 獲取異步任務(wù)結(jié)果
        results = [r.get() for r in results]
        print(f"Results: {results}")

在這個(gè)例子中，我們使用 apply_async() 異步執(zhí)行任務(wù)，而主進(jìn)程在等待任務(wù)完成之前可以執(zhí)行其他操作。最終我們通過 get() 方法獲取每個(gè)任務(wù)的結(jié)果。

5.2 異常處理

在并發(fā)編程中，處理異常是非常重要的。如果某個(gè)進(jìn)程發(fā)生異常，我們需要確保能夠捕捉到這些異常，并做出相應(yīng)的處理。apply_async() 提供了 error_callback 參數(shù)，可以用于捕捉異步任務(wù)中的異常。

def worker(num):
    if num == 3:
        raise ValueError("模擬錯(cuò)誤")
    return num * 2
def handle_error(e):
    print(f"捕獲異常: {e}")
if __name__ == '__main__':
    with multiprocessing.Pool(processes=4) as pool:
        results = [pool.apply_async(worker, args=(i,), error_callback=handle_error) for i in range(10)]
        for result in results:
            try:
                print(result.get())
            except Exception as e:
                print(f"主進(jìn)程捕獲異常: {e}")

在這個(gè)例子中，如果某個(gè)任務(wù)拋出了異常，error_callback 函數(shù)會(huì)捕捉到，并進(jìn)行處理。

六、實(shí)際應(yīng)用場(chǎng)景

6.1 CPU 密集型任務(wù)

多進(jìn)程并行處理非常適合處理 CPU 密集型任務(wù)，如圖像處理、大規(guī)模數(shù)據(jù)運(yùn)算等。在這些任務(wù)中，計(jì)算量非常大，多個(gè)進(jìn)程可以同時(shí)利用系統(tǒng)的多個(gè) CPU 核心，顯著縮短處理時(shí)間。

def cpu_intensive_task(n):
    total = 0
    for i in range(10**6):
        total +=
 i * n
    return total
if __name__ == '__main__':
    with multiprocessing.Pool(processes=4) as pool:
        results = pool.map(cpu_intensive_task, range(10))
        print(results)

6.2 IO 密集型任務(wù)

對(duì)于 IO 密集型任務(wù)，如網(wǎng)絡(luò)請(qǐng)求、文件讀寫等，由于進(jìn)程大部分時(shí)間在等待外部資源響應(yīng)，所以進(jìn)程間的并發(fā)性能提升可能沒有 CPU 密集型任務(wù)明顯。但仍然可以通過多進(jìn)程方式提高并發(fā)度，減少等待時(shí)間。

七、總結(jié)

通過本文的學(xué)習(xí)，我們了解了如何使用 Python 中的 multiprocessing 模塊創(chuàng)建進(jìn)程池，并將任務(wù)分配給多個(gè)進(jìn)程執(zhí)行。進(jìn)程池的使用可以幫助我們有效管理并發(fā)任務(wù)，提高程序執(zhí)行效率，尤其是在處理 CPU 密集型任務(wù)時(shí)效果顯著。

在實(shí)踐中，使用進(jìn)程池時(shí)我們還需要注意以下幾點(diǎn)：

• 資源管理：確保合理使用進(jìn)程池，避免創(chuàng)建過多進(jìn)程導(dǎo)致系統(tǒng)資源不足。
• 任務(wù)分配：根據(jù)任務(wù)的不同類型（如 CPU 密集型和 IO 密集型），選擇合適的并行處理方法。
• 異常處理：在多進(jìn)程環(huán)境中捕捉和處理異常，避免因?yàn)閱蝹€(gè)進(jìn)程出錯(cuò)而導(dǎo)致整個(gè)程序崩潰。

通過掌握這些技巧，你可以在 Python 編程中充分利用并行處理的優(yōu)勢(shì)，構(gòu)建更加高效的應(yīng)用程序。

到此這篇關(guān)于Python 如何使用 multiprocessing 模塊創(chuàng)建進(jìn)程池的文章就介紹到這了,更多相關(guān)Python multiprocessing 模塊創(chuàng)建進(jìn)程池內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論