Python多进程调用时如何避免列表在函数间共享

问题描述 #

我用multiprocessing将函数加入队列做并行处理，函数接收两个列表作为输入。循环执行6次操作时，每次把循环计数器追加到列表末尾，却发现列表共享了内存指针——每次函数执行时列表都会变长，而非成为每次执行的独立局部变量。我想让每个列表成为独立变量，但没找到合适方案，只搜到创建共享变量的相关内容。

另外，有没有更优的并行实现方式？我不太擅长写脚本，也从未尝试过并行处理，目前都是复制粘贴多进程相关代码片段。

背景：我正在测试Risk游戏所有可能的骰子组合，计算进攻方或防守方的获胜概率。不同于常规的“进攻方掷3枚骰子、防守方掷2枚骰子”，我需要测试双方使用更多骰子的情况，因此引入并行处理以应对额外骰子带来的指数级耗时增长。

测试用简化代码片段 #

#!/usr/bin/python3.5
#good break down of logic for traditional 3v2 in here: http://www.datagenetics.com/blog/november22011/index.html
#multithreading logic from: https://superfastpython.com/parallel-nested-for-loops-in-python/

import threading
import multiprocessing.pool
import queue

def compare(attack, defence):
    aWon = 0
    print(defence)
    for i in range(len(defence)):
        if attack[i] > defence[i]:
            aWon += 1
    aWins[aWon] += 1 #aWins is a shared global array to keep track of how many attacker die won a given encounter
 
def d1(attack, defence):
    global queue
    defence.sort(reverse=True)
    for i in range(1, 7):
        defence.append(i+10)
        queue.put((compare(attack, defence), (i,)))

aWins = [0,0,0,0,0,0] #0 index incremented when attacker won 0 die in a set; 1 index when they won 1, etc
global queue
queue = queue.Queue()
queue.put((d1([1,1,1,1,1,1], [1,1])))
 
total_tasks = 216
with multiprocessing.pool.ThreadPool(total_tasks) as pool:
    for _ in range(total_tasks):
        task, args = queue.get()
        pool.apply_async(task, args)
    pool.close()
    pool.join()

调试打印输出 #

[1, 1, 11]
[1, 1, 11, 12]
[1, 1, 11, 12, 13]
[1, 1, 11, 12, 13, 14]
[1, 1, 11, 12, 13, 14, 15]

期望的调试打印输出 #

[1, 1, 11]
[1, 1, 12]
[1, 1, 13]
[1, 1, 14]
[1, 1, 15]

解决方案 #

一、解决列表共享问题 #

问题核心：d1函数中一直在修改同一个defence列表对象（列表是可变类型，append会直接修改原对象），导致后续任务拿到的都是被修改过的列表。要创建独立列表，需每次循环时复制原列表再追加元素：

修改后的d1函数及相关代码：

def d1(attack, defence):
    global queue
    # 先排序得到新列表，避免修改原输入
    defence_sorted = sorted(defence, reverse=True)
    for i in range(1, 7):
        # 每次循环复制排序后的列表，生成独立的新列表
        new_defence = defence_sorted.copy()
        new_defence.append(i+10)
        # 修正队列存储格式：函数 + (参数元组)
        queue.put((compare, (attack, new_defence)))

# 替换原队列放入逻辑：直接调用d1填充队列，而非把d1放入队列
d1([1,1,1,1,1,1], [1,1])

此外，全局变量aWins在多线程中会有竞争问题（多个线程同时修改会导致计数错误），需添加线程锁保护：

import threading

aWins = [0,0,0,0,0,0]
aWins_lock = threading.Lock()

def compare(attack, defence):
    aWon = 0
    print(defence)
    # 修正比较逻辑：取双方数量较少的骰子数进行比较（符合Risk规则）
    min_dice_count = min(len(attack), len(defence))
    # 对攻防骰子降序排序后取前min_dice_count个比较
    for a, d in zip(sorted(attack, reverse=True)[:min_dice_count], 
                    sorted(defence, reverse=True)[:min_dice_count]):
        if a > d:
            aWon += 1
    # 加锁修改共享变量
    with aWins_lock:
        aWins[aWon] += 1

二、更优的并行实现方式 #

你的任务属于CPU密集型，多线程受GIL锁限制效率不高，推荐用多进程池+组合生成的方式，逻辑更清晰且效率更高：

import multiprocessing
from itertools import product

def calculate_single_combination(attack_dice, defence_dice):
    """计算单个攻防骰子组合的进攻方获胜骰子数"""
    min_count = min(len(attack_dice), len(defence_dice))
    a_won = 0
    # 按Risk规则排序后比较
    for a, d in zip(sorted(attack_dice, reverse=True)[:min_count], 
                    sorted(defence_dice, reverse=True)[:min_count]):
        if a > d:
            a_won += 1
    return a_won

def generate_all_dice_combinations(attack_num, defence_num):
    """生成所有可能的攻防骰子组合"""
    # 生成进攻方所有骰子组合（每个骰子1-6）
    attack_combs = product(range(1,7), repeat=attack_num)
    # 生成防守方所有骰子组合
    defence_combs = product(range(1,7), repeat=defence_num)
    # 生成所有组合对
    return [(list(a), list(d)) for a in attack_combs for d in defence_combs]

if __name__ == "__main__":
    # 配置攻防骰子数量
    ATTACK_DICE_NUM = 6
    DEFENCE_DICE_NUM = 2
    
    # 生成所有组合
    all_combinations = generate_all_dice_combinations(ATTACK_DICE_NUM, DEFENCE_DICE_NUM)
    
    # 用多进程池处理任务
    with multiprocessing.Pool() as pool:
        # starmap自动拆分元组参数传入函数
        results = pool.starmap(calculate_single_combination, all_combinations)
    
    # 统计结果
    max_win_count = min(ATTACK_DICE_NUM, DEFENCE_DICE_NUM)
    aWins = [0]*(max_win_count + 1)
    for res in results:
        aWins[res] += 1
    
    # 输出概率
    total = sum(aWins)
    print("进攻方获胜骰子数统计：")
    for win_count, count in enumerate(aWins):
        probability = count / total * 100
        print(f"获胜{win_count}个骰子：{count}次，概率：{probability:.2f}%")

该方案优势：

1. 用itertools.product直接生成所有组合，逻辑清晰，避免手动管理队列的麻烦
2. 多进程池自动分配任务、收集结果，无需手动处理线程/进程同步
3. 每个任务处理独立的组合，无共享变量冲突问题
4. 代码结构模块化，容易扩展到不同骰子数量的测试场景

额外优化建议：

• 如果组合数量过大（比如6攻6守有6^12≈21亿种组合），可以分批次处理，避免内存占用过高
• 用pool.imap_unordered替代starmap，可实时获取并处理结果，无需等待所有任务完成

内容的提问来源于stack exchange，提问作者Shahad