如何拆分列表为多批次分配给线程Worker，实现平方和并行计算

完整实现代码 #

1. 单线程方案 #

直接遍历计算所有元素的平方和，用time.perf_counter()统计耗时：

import time

def single_thread_calculate():
    total = sum(x*x for x in range(2001))
    return total

if __name__ == "__main__":
    # 单线程计时
    start_time = time.perf_counter()
    single_result = single_thread_calculate()
    single_duration = time.perf_counter() - start_time
    print(f"单线程结果：{single_result}，耗时：{single_duration:.6f}秒")

2. 多线程并行方案 #

拆分列表为多个子批次，用线程池分配任务，每个线程负责计算一个子批次的平方和，最后汇总结果：

import time
from concurrent.futures import ThreadPoolExecutor

# 完善后的线程任务函数：接收子列表，返回该子列表的平方和
def calculate_sub_sum(sub_list):
    return sum(x*x for x in sub_list)

def multi_thread_calculate(total_elements=2001, thread_num=4):
    # 拆分列表为指定数量的批次
    step = total_elements // thread_num
    batches = []
    for i in range(thread_num):
        start = i * step
        # 最后一个批次兜底处理剩余元素，避免遗漏
        end = start + step if i != thread_num-1 else total_elements
        batches.append(range(start, end))
    
    # 用线程池批量执行任务
    with ThreadPoolExecutor(max_workers=thread_num) as executor:
        sub_results = list(executor.map(calculate_sub_sum, batches))
    
    # 汇总所有子批次的计算结果
    return sum(sub_results)

if __name__ == "__main__":
    # 多线程计时
    start_time = time.perf_counter()
    multi_result = multi_thread_calculate()
    multi_duration = time.perf_counter() - start_time
    print(f"多线程结果：{multi_result}，耗时：{multi_duration:.6f}秒")
    
    # 验证两种方案结果一致性
    assert single_result == multi_result, "两种方案计算结果不一致！"
    print("结果验证通过：单线程与多线程计算结果完全一致")

关键说明 #

• calculate_sub_sum函数职责单一，仅处理子列表的平方和计算，适配线程任务的设计需求
• 列表拆分时通过最后一个批次兜底剩余元素，确保所有元素都被计算
• 使用ThreadPoolExecutor简化线程生命周期管理，无需手动创建、启动和回收线程
• 通过assert校验结果一致性，避免并行逻辑导致的计算错误

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