【💊 Python的解憂錦囊】如何在multithread/multiprocess傳遞固定參數？

撰寫Python的朋友都知道multithread/multiprocess能為我們帶來效能的改進，減少硬體資源的閒置，但在撰寫的過程中常常會發現到我們所設計的工作池模式會需要將「待辦清單」的工作項目當成參數傳遞進去執行， 除了「待辦清單」之外， 其餘的參數基本上都是固定的， 基於這樣的需求之下， 我們要怎麼完成呢？ 讓我們耐心的看完這個篇章。

我們通常會這樣做…

假設我們設計了一個工作但尚未實作工作詳細內容， 僅印出工作資訊如下 :

def job(name: str, action: str, item: str):
    """工作內容
    
    Args : 
        name (str): 什麼樣的工作
        action (str): 工作的行為(加工、蓋房、...)
        item: 工作的項目
    Retruns:
        None
    """
    result = f'{name} 正在 {action} {item}'
    return result

接著我們在主程序設計好我們要執行「什麼樣的工作」、「工作的行為」， 接著我們會有許多的「待辦事項」需要執行， 接著招聘好工人(num_workers)之後就可以根據作業區擴展廠區(pool)，每個作業區獨立運作這些待辦清單， 那我們可能會這樣撰寫程式：

import concurrent.futures

if __name__ == "__main__":
    # 設計今天的主題
    name = '食品加工廠'
    action = '製作'
    
    # 待辦事項
    todo_list = ['熱狗', '炸雞', '薯條', '肉乾']
    
    num_workers = 3
    
    # 創建一個多進程池，根據上述的工人數量擴展工作池
    with concurrent.futures.ProcessPoolExecutor(max_workers=num_workers) as executor:

				# 我們將大量的待辦事項轉換成每個worker需要執行的
        params = [(name, action, item) for item in todo_list]
        results = executor.map(job, params)
    
    # 印出執行結果
    for result in results:
        print(result)

但上述的作法真的好嗎？

我們的params不會隨著「待辦清單」越多(上千萬個事項)， 導致記憶體爆掉嗎？ 不妨來看看我們更改後的範例， 假設有1000萬的「待辦清單」時會發生什麼狀況？ 我們也順便埋入執行的估測時間來實際看看結果。

import multiprocessing
import time
import random
import sys

def job(name: str, action: str, item: str):
    """工作內容
    
    Args : 
        name (str): 工作名稱
        action (str): 動作
        item: 工作的項目
    Retruns:
        None
    """
    secs = random.random()
    result = f'{name} 正在 {action} {item} 花費了 {secs} 秒'
    time.sleep(secs)
    return result

if __name__ == "__main__":
    # 固定的參數
    name = '食品加工廠'
    action = '製作'
    
    # 主程序提供的待辦事項清單
    # todo_list = ['熱狗', '炸雞', '薯條', '肉乾']
    todo_list = [f"工作{i}" for i in range(1, 10000000)]
    
    num_workers = 3
    
    # 創建一個多進程池，這裡使用3個進程
    with multiprocessing.Pool(processes=num_workers) as pool:
        start_time = time.time()
        params = [(name, action, item) for item in todo_list]
        end_time = time.time()
        
        use_bytes = sys.getsizeof(params)
        use_mb = use_bytes / 1048576
        print(f'渲染參數花費的時間: {end_time - start_time} 秒, 耗用的記憶體: {use_mb} MB')
        
        results = pool.starmap(job, params)
        for result in results:
            print(result)

我們會發現以下光是簡單的參數就花費如此之多的記憶體耗用量， 那面對大數據時怎麼辦？

因此我們可以這樣做…

我們在「【Python 軍火庫🧨 - functools】使用partial來設計函數樣板」有介紹到「functools.partial」這個工具庫， 我們可以利用partial的技巧製造出固定參數的新函式， 以不變應萬變，套用到multiprocess之前就不需要一堆複製的資源耗費…。

那在進入主題之前， 我們先來複習一下關於functools.partial函式， 他可以幫我們製作出固定參數的樣版。

但聰明的大家有沒有觀察到一個狀況， 那就是動態的參數通常在前面(a, b)， 而固定的參數放在後段(c)， 因此底下我們的job的參數設計勢必要改一改， 由於我們的item會隨著todolist而變化， 因此需要將函式參數順序稍微修改一下成「job(item: str, name: str, action: str)」。

另外在於multiprocess的部份， 我們原先使用的是「starmap」接受多參數的模式， 但經上述演示之後覺得對於大數據的處理不太妥當， 因此我們可以更換成「map(func, iterable[, chunksize])」， 他們的差異主要在於

import multiprocessing
import time
import random
import sys
from functools import partial

def job(item: str, name: str, action: str):
    """工作內容
    
    Args : 
        item (str): 工作的項目 [可變]
        name (str): 工作名稱 [固定]
        action (str): 動作 [固定]
        
    Retruns:
        None
    """
    secs = random.random()
    result = f'{name} 正在 {action} {item} 花費了 {secs} 秒'
    return result

if __name__ == "__main__":
    # 固定的參數
    name = '食品加工廠'
    action = '製作'
    
    # 主程序提供的待辦事項清單
    # todo_list = ['熱狗', '炸雞', '薯條', '肉乾']
    todo_list = [f"工作{i}" for i in range(1, 10000000)]
    
    num_workers = 24
    
    # 創建一個多進程池，這裡使用3個進程
    with multiprocessing.Pool(processes=num_workers) as pool:
        start_time = time.time()
        job_func = partial(job, name=name, action=action)
        end_time = time.time()
        
        use_bytes = sys.getsizeof(job_func)
        use_mb = use_bytes / 1048576
        print(f'渲染參數花費的時間: {end_time - start_time} 秒, 耗用的記憶體: {use_mb} MB')
        
        results = pool.map(job_func, todo_list)
        for result in results:
            print(result)

乍看之下是不是會誤以為花費更久的時間了呢？ 請仔細看一下它的數字是否怪怪的， 有負數的出現…， 這是科學記號， 代表是一個非常小的數字， 也就是幾乎沒有耗損的過程， 這面對於大數據的處理， 相對能夠帶來極大的效能增幅。

結語

上述的例子仍有一些缺陷， 雖然實際上仍是以multiprocess的方式並行處理我們的運算， 但必須等待全部的工作完成後， 才回到主線程， 這對於我們跟蹤程式進度的應用會稍微不利， 不過沒關係， 我們後續也會針對這個部份進行重點分享， 歡迎持續追蹤， 讓我們一起探究軟體開發的大小事。

學習軟體開發的路上常常苦於網路資訊爆炸嗎？ 教學何其多，但卻遇到無法明確選擇的困境呢？ 歡迎加入「🔒 阿Han的軟體心法實戰營」， 這裡不給您冗餘的雜訊， 單刀直入直接送您業界開發重點， 避開選擇障礙的困境， 讓您獲得業界標準的開發起手式， 成為Top 1的頂尖人才。