快捷導(dǎo)航

Python使用線程池傳遞多個(gè)參數(shù)的幾種方法

更新時(shí)間：2026年04月14日 09:34:14 作者：獨(dú)隅

本文介紹了在Python中使用線程池傳遞多個(gè)參數(shù)的四種方法,并推薦使用stmap方式,這種方法代碼清晰易讀,參數(shù)傳遞靈活,易于維護(hù)和擴(kuò)展,適用于為每個(gè)設(shè)備指定不同參數(shù)的場(chǎng)景,需要的朋友可以參考下

方法一：使用zip和*解包

from concurrent.futures import ThreadPoolExecutor

def worker(phone, x, y, duration):
    print(f"設(shè)備: {phone}, 位置: ({x}, {y}), 時(shí)長(zhǎng): {duration}")
    # 這里執(zhí)行實(shí)際的投屏操作

def thread_pool_with_zip():
    phones = ["172.26.101.164", "172.26.101.112", "172.26.101.113"]
    x_positions = [100, 500, 900]  # X坐標(biāo)
    y_positions = [100, 100, 100]  # Y坐標(biāo)
    durations = [20, 25, 30]  # 投屏?xí)r長(zhǎng)
    
    with ThreadPoolExecutor(max_workers=3) as executor:
        # 使用zip打包參數(shù)，然后解包傳遞給worker函數(shù)
        executor.map(worker, phones, x_positions, y_positions, durations)

方法二：使用partial函數(shù)固定部分參數(shù)

from concurrent.futures import ThreadPoolExecutor
from functools import partial

def worker(phone, x, y, duration):
    print(f"設(shè)備: {phone}, 位置: ({x}, {y}), 時(shí)長(zhǎng): {duration}")

def thread_pool_with_partial():
    phones = ["172.26.101.164", "172.26.101.112", "172.26.101.113"]
    
    # 固定x, y, duration參數(shù)
    fixed_worker = partial(worker, x=100, y=100, duration=20)
    
    with ThreadPoolExecutor(max_workers=3) as executor:
        executor.map(fixed_worker, phones)

方法三：使用字典或元組包裝參數(shù)

from concurrent.futures import ThreadPoolExecutor

def worker(params):
    phone = params['phone']
    x = params['x']
    y = params['y']
    duration = params['duration']
    print(f"設(shè)備: {phone}, 位置: ({x}, {y}), 時(shí)長(zhǎng): {duration}")

def thread_pool_with_dict():
    # 參數(shù)列表
    params_list = [
        {'phone': "172.26.101.164", 'x': 100, 'y': 100, 'duration': 20},
        {'phone': "172.26.101.112", 'x': 500, 'y': 100, 'duration': 25},
        {'phone': "172.26.101.113", 'x': 900, 'y': 100, 'duration': 30}
    ]
    
    with ThreadPoolExecutor(max_workers=3) as executor:
        executor.map(worker, params_list)

方法四：使用starmap（推薦）

from concurrent.futures import ThreadPoolExecutor

def worker(phone, x, y, duration):
    print(f"設(shè)備: {phone}, 位置: ({x}, {y}), 時(shí)長(zhǎng): {duration}")

def thread_pool_with_starmap():
    # 參數(shù)元組列表
    args_list = [
        ("172.26.101.164", 100, 100, 20),
        ("172.26.101.112", 500, 100, 25),
        ("172.26.101.113", 900, 100, 30)
    ]
    
    with ThreadPoolExecutor(max_workers=3) as executor:
        # 使用starmap解包元組參數(shù)
        executor.map(lambda args: worker(*args), args_list)

在您的代碼中應(yīng)用

修改您的代碼以支持多參數(shù)傳遞：

def Scrcpy(params):
    """接收參數(shù)字典的投屏函數(shù)"""
    phone = params['phone']
    x = params.get('x', 100)  # 默認(rèn)值
    y = params.get('y', 100)  # 默認(rèn)值
    duration = params.get('duration', 20)  # 默認(rèn)值
    
    RestartScrcpy(phone)
    time.sleep(5)
    phone_num = Phone_number(phone)
    
    try:
        _IP_CONNEXT = f"{phone}:12324"
        run_cmd(f"adb connect {_IP_CONNEXT}")  
        # 添加窗口位置參數(shù)
        run_cmd(f'scrcpy -s {_IP_CONNEXT} --time-limit {duration} --max-size=1280 --window-title={phone_num} --window-x={x} --window-y={y}') 
        run_cmd(f"adb disconnect {_IP_CONNEXT}")
    except Exception as e:
        log(logging.ERROR, f"Scrcpy: {e}")

def thread_pool_multiple_params():
    # 定義設(shè)備及其參數(shù)
    device_configs = [
        {'phone': "172.26.101.164", 'x': 100, 'y': 100, 'duration': 20},
        {'phone': "172.26.101.112", 'x': 800, 'y': 100, 'duration': 25},
        # 添加更多設(shè)備...
    ]
    
    try:
        with ThreadPoolExecutor(max_workers=len(device_configs)) as executor:
            executor.map(Scrcpy, device_configs)
    except Exception as e:
        log(logging.ERROR, f"線程池異常: {e}")

# 在主函數(shù)中使用
if __name__ == '__main__':
    while True:
        if len(PHONES) >= 2:
            selected = random.sample(PHONES, 2)
            log(logging.DEBUG, f"--> 隨機(jī)選取的投屏設(shè)備：{selected}")
            
            # 為每個(gè)設(shè)備創(chuàng)建參數(shù)配置
            configs = []
            for i, phone in enumerate(selected):
                configs.append({
                    'phone': phone,
                    'x': 100 + i * 700,  # 水平排列窗口
                    'y': 100,
                    'duration': 20
                })
            
            thread_pool_multiple_params()
            time.sleep(5)
        # ... 其他條件判斷

使用進(jìn)程池（multiprocessing.Pool）的多參數(shù)傳遞

import multiprocessing

def worker(args):
    phone, x, y, duration = args
    print(f"設(shè)備: {phone}, 位置: ({x}, {y}), 時(shí)長(zhǎng): {duration}")

def multiprocessing_pool_multiple_params():
    args_list = [
        ("172.26.101.164", 100, 100, 20),
        ("172.26.101.112", 500, 100, 25),
        ("172.26.101.113", 900, 100, 30)
    ]
    
    with multiprocessing.Pool(processes=3) as pool:
        pool.map(worker, args_list)