Python使用多進程執(zhí)行同一個函數(shù)的多種方法

更新時間：2026年04月15日 09:25:04 作者：獨隅

本文介紹了在Python中使用多進程執(zhí)行相同函數(shù)的三種方法：Process類、Pool進程池和apply_async異步執(zhí)行,比較了各種方法的適用場景、優(yōu)缺點,并提供了實用技巧和注意事項,推薦使用Pool進程池處理大量任務,它簡單易用且自動負載均衡,需要的朋友可以參考下

方法一：使用 Process 類創(chuàng)建獨立進程

這種方法適合需要精確控制每個進程的場景。

import multiprocessing
import time

def worker(task_name, duration):
    print(f"進程 {task_name} 開始執(zhí)行，預計耗時 {duration} 秒")
    time.sleep(duration)
    print(f"進程 {task_name} 執(zhí)行完成")
    return f"{task_name} 處理結果"

if __name__ == "__main__":
    processes = []
    tasks = [("任務A", 2), ("任務B", 3), ("任務C", 1)]
    
    # 創(chuàng)建并啟動多個進程
    for name, duration in tasks:
        p = multiprocessing.Process(target=worker, args=(name, duration))
        processes.append(p)
        p.start()
        print(f"已啟動進程: {name}")
    
    # 等待所有進程完成
    for p in processes:
        p.join()
    
    print("所有進程執(zhí)行完畢")

方法二：使用 Pool 進程池（推薦）

進程池更適合處理大量任務，可以自動管理進程的創(chuàng)建和回收。

import multiprocessing
import os

def worker(x):
    print(f"進程 {os.getpid()} 正在處理數(shù)據(jù): {x}")
    result = x * x  # 模擬計算任務
    return result

if __name__ == "__main__":
    # 創(chuàng)建進程池，進程數(shù)通常設為CPU核心數(shù)
    with multiprocessing.Pool(processes=multiprocessing.cpu_count()) as pool:
        data = [1, 2, 3, 4, 5, 6, 7, 8]
        
        # 方法1: map - 阻塞式，按順序返回結果
        results = pool.map(worker, data)
        print(f"map 結果: {results}")
        
        # 方法2: map_async - 非阻塞式
        async_result = pool.map_async(worker, data)
        results_async = async_result.get()  # 獲取結果時會阻塞
        print(f"map_async 結果: {results_async}")

方法三：使用 apply_async 實現(xiàn)異步執(zhí)行

適合需要更靈活控制任務提交和結果獲取的場景。

import multiprocessing
import time
import random

def worker(task_id):
    sleep_time = random.uniform(0.5, 2.0)
    time.sleep(sleep_time)
    return f"任務{task_id} 完成，耗時{sleep_time:.2f}秒"

if __name__ == "__main__":
    with multiprocessing.Pool(4) as pool:
        # 提交多個異步任務
        async_results = []
        for i in range(6):
            result = pool.apply_async(worker, args=(i,))
            async_results.append(result)
        
        print("所有任務已提交，主進程可繼續(xù)執(zhí)行其他工作...")
        
        # 按完成順序獲取結果（無序）
        for result in async_results:
            print(result.get())  # 阻塞直到有結果可用

方法對比與選擇指南

方法	適用場景	優(yōu)點	缺點
Process	進程數(shù)少，需要精細控制	控制靈活，可設置不同參數(shù)	需要手動管理進程生命周期
Pool.map	批量處理相同任務	簡單易用，自動負載均衡	必須等待所有任務完成
Pool.apply_async	異步任務處理	非阻塞，效率高	結果獲取稍復雜

實用技巧與注意事項

進程間通信：使用 Queue 進行數(shù)據(jù)傳遞

def worker(queue):
    data = queue.get()
    processed = f"處理后的: {data}"
    queue.put(processed)

錯誤處理：使用回調函數(shù)處理異常

def error_callback(error):
    print(f"任務執(zhí)行出錯: {error}")

pool.apply_async(worker, args=(data,), error_callback=error_callback)

性能優(yōu)化：合理設置進程數(shù)和 chunksize

# 根據(jù)任務類型調整參數(shù)
results = pool.map(worker, data, chunksize=len(data)//multiprocessing.cpu_count())

實際應用示例：并行下載任務

import multiprocessing
import requests

def download_url(url):
    try:
        response = requests.get(url, timeout=10)
        return f"{url}: 成功, 大小 {len(response.content)} bytes"
    except Exception as e:
        return f"{url}: 失敗, 錯誤 {str(e)}"

if __name__ == "__main__":
    urls = [
        "https://httpbin.org/delay/1",
        "https://httpbin.org/delay/2", 
        "https://httpbin.org/delay/1"
    ]
    
    with multiprocessing.Pool(3) as pool:
        results = pool.map(download_url, urls)
    
    for result in results:
        print(result)

根據(jù)你的具體需求選擇合適的方案，Pool 進程池 在大多數(shù)情況下都是最實用和高效的選擇。

以上就是Python使用多進程執(zhí)行同一個函數(shù)的多種方法的詳細內(nèi)容，更多關于Python多進程執(zhí)行同一個函數(shù)的資料請關注腳本之家其它相關文章！

您可能感興趣的文章: