快捷導(dǎo)航

python并行設(shè)計的實現(xiàn)

更新時間：2024年09月23日 09:42:50 作者：pumpkin84514

python中的并行設(shè)計可以顯著增強程序處理大量數(shù)據(jù)或復(fù)雜計算的速度,通過使用threading、multiprocessing和concurrent.futures等庫,開發(fā)者可以有效利用多核CPU的計算力,下面就來詳細的介紹一下

并行設(shè)計簡介

并行設(shè)計指的是同時運行多個計算任務(wù)，這樣可以充分利用多核CPU的計算能力。Python中常用的并行編程庫包括threading、multiprocessing和concurrent.futures。

1. 使用 threading 模塊

threading 模塊提供了一個基于線程的并行執(zhí)行方式。線程是輕量級的，并且共享相同的內(nèi)存空間，所以線程間通信開銷較小。下面是一個簡單的例子：

import threading
import time

def worker(num):
    """線程工作函數(shù)"""
    print(f"線程 {num} 開始")
    time.sleep(2)
    print(f"線程 {num} 結(jié)束")

threads = []

for i in range(5):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print("所有線程已完成")

在這個例子中，我們創(chuàng)建了5個線程，每個線程執(zhí)行 worker 函數(shù)并等待2秒。最后，我們使用 join() 方法確保主線程等待所有子線程完成。

2. 使用 multiprocessing 模塊

multiprocessing 模塊提供了基于進程的并行執(zhí)行方式。進程獨立于其他進程，每個進程有自己的內(nèi)存空間，所以進程間通信開銷較大，但可以利用多核CPU的全部能力。下面是一個例子：

from multiprocessing import Process
import time

def worker(num):
    """進程工作函數(shù)"""
    print(f"進程 {num} 開始")
    time.sleep(2)
    print(f"進程 {num} 結(jié)束")

processes = []

for i in range(5):
    p = Process(target=worker, args=(i,))
    processes.append(p)
    p.start()

for p in processes:
    p.join()

print("所有進程已完成")

這個例子與線程的例子類似，但使用的是進程。每個進程獨立運行，并在完成后返回。

3. 使用 concurrent.futures 模塊

concurrent.futures 模塊提供了一個高級接口，用于管理線程池和進程池。它的使用更加簡便，推薦在實際項目中使用。下面是一個使用線程池的例子：

from concurrent.futures import ThreadPoolExecutor
import time

def worker(num):
    print(f"線程 {num} 開始")
    time.sleep(2)
    print(f"線程 {num} 結(jié)束")

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(worker, i) for i in range(5)]
    for future in futures:
        future.result()

print("所有線程已完成")

以及使用進程池的例子：

from concurrent.futures import ProcessPoolExecutor
import time

def worker(num):
    print(f"進程 {num} 開始")
    time.sleep(2)
    print(f"進程 {num} 結(jié)束")

with ProcessPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(worker, i) for i in range(5)]
    for future in futures:
        future.result()

print("所有進程已完成")

邏輯和應(yīng)用場景

線程適用于IO密集型任務(wù)，例如網(wǎng)絡(luò)請求、文件讀取等。這些任務(wù)在等待IO操作完成時，CPU可以執(zhí)行其他任務(wù)，從而提高效率。
進程適用于CPU密集型任務(wù)，例如復(fù)雜計算、圖像處理等。這些任務(wù)需要大量計算資源，使用多進程可以充分利用多核CPU的能力。
concurrent.futures 模塊的線程池和進程池適合管理大量并發(fā)任務(wù)，簡化代碼并提高可讀性。

實際應(yīng)用示例

假設(shè)我們有一個需要處理大量數(shù)據(jù)的場景，例如對一組圖像進行處理。我們可以使用 concurrent.futures 模塊來實現(xiàn)并行處理：

from concurrent.futures import ProcessPoolExecutor
from PIL import Image
import os

def process_image(image_path):
    img = Image.open(image_path)
    img = img.convert("L")  # 轉(zhuǎn)換為灰度圖
    img.save(f"processed/{os.path.basename(image_path)}")
    print(f"{image_path} 已處理")

image_paths = ["image1.jpg", "image2.jpg", "image3.jpg"]

if not os.path.exists("processed"):
    os.makedirs("processed")

with ProcessPoolExecutor(max_workers=4) as executor:
    executor.map(process_image, image_paths)

print("所有圖像已處理")

在這個例子中，我們使用多進程池并行處理圖像，顯著提高了處理效率。