如何用Python多线程分批处理列表元素且无线程重叠？

需求说明 #

你需要处理一个包含50000个元素的字符串列表，要求按批次处理，每批次内的元素并行执行字符串转整数操作，且必须等待当前批次所有元素处理完成后，再开始下一批次的处理。

基础线程实现方案（手动管理线程） #

使用threading.Thread手动创建和管理线程，通过join()方法等待批次内所有线程完成：

import threading
from typing import List

def to_int(s: str) -> int:
    num = int(s)
    print(num)
    return num

def process_batch(batch: List[str]):
    # 存储当前批次的所有线程对象
    threads = []
    # 为每个元素创建线程并启动
    for s in batch:
        thread = threading.Thread(target=to_int, args=(s,))
        threads.append(thread)
        thread.start()
    # 等待当前批次所有线程执行完毕
    for thread in threads:
        thread.join()

if __name__ == "__main__":
    # 模拟50000个元素的大型列表
    large_list = [str(i) for i in range(1, 50001)]
    batch_size = 3  # 可根据实际需求调整批次大小

    # 分批遍历列表
    for i in range(0, len(large_list), batch_size):
        current_batch = large_list[i:i+batch_size]
        print(f"开始处理批次: {current_batch}")
        process_batch(current_batch)
        print("当前批次处理完成\n")

关键逻辑说明

• process_batch函数负责单批次的并行处理：先为每个元素创建线程并启动，再通过join()逐个等待线程结束——主线程会阻塞直到该批次所有线程都执行完成，确保批次间的顺序执行。
• 批次大小可根据系统性能调整，避免单次创建过多线程导致资源占用过高。

简化实现方案（使用线程池） #

使用concurrent.futures.ThreadPoolExecutor可以更简洁地管理线程，无需手动创建和跟踪线程对象：

from concurrent.futures import ThreadPoolExecutor
from typing import List

def to_int(s: str) -> int:
    num = int(s)
    print(num)
    return num

if __name__ == "__main__":
    large_list = [str(i) for i in range(1, 50001)]
    batch_size = 3

    for i in range(0, len(large_list), batch_size):
        current_batch = large_list[i:i+batch_size]
        print(f"开始处理批次: {current_batch}")
        # 使用线程池执行当前批次任务，with块结束时自动等待所有任务完成
        with ThreadPoolExecutor(max_workers=len(current_batch)) as executor:
            executor.map(to_int, current_batch)
        print("当前批次处理完成\n")

关键逻辑说明

• ThreadPoolExecutor的上下文管理器（with块）会自动处理线程的创建、执行和销毁：当退出with块时，会等待所有提交的任务执行完毕，天然满足批次间的顺序要求。
• executor.map方法会自动将批次中的每个元素传入to_int函数，实现并行执行。如果批次过大，建议将max_workers设置为固定值（如CPU核心数的2倍），避免线程过多导致系统负载过高。

注意事项 #

• 字符串转整数属于CPU密集型任务，Python的GIL（全局解释器锁）会限制多线程的真正并行能力，如果追求更高效率，可考虑使用多进程（multiprocessing模块）。
• 若需要收集每个线程的返回值，可通过executor.map的返回值或threading.Thread结合队列（queue.Queue）实现。

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