如何让ThreadPoolExecutor在获取指定数量结果后停止执行

你遇到的问题其实是ThreadPoolExecutor.map()的工作机制导致的——这个方法会一次性把所有buckets对应的任务提交到线程池，哪怕你在循环里写了break，已经提交的任务还是会继续执行，你只是停止把后续结果加到res_list里而已，根本没法真正“终止循环”或者停止任务。

下面给你两种可行的解决方案，根据你的需求选择：

方案1：停止接收结果并取消未执行的任务 #

如果你只需要在res_list达到目标长度后，停止收集结果，同时取消还没开始运行的任务（已经在跑的任务会继续执行完），可以用submit结合as_completed来实现：

from concurrent.futures import ThreadPoolExecutor, as_completed

res_list = []
# 根据你的条件计算目标长度：对应你原来的 len(res_list) - bucket_size < number_reviews
target_length = number_reviews + bucket_size

with ThreadPoolExecutor() as pool:
    # 先提交所有任务，保存每个任务的future对象
    futures = [pool.submit(load_bucket, bucket) for bucket in buckets]
    
    # 遍历已经完成的任务
    for future in as_completed(futures):
        if len(res_list) < target_length:
            # 获取任务结果并加入列表
            result = future.result()
            res_list.append(result)
        else:
            # 取消所有还未完成的任务
            for remaining_future in futures:
                if not remaining_future.done():
                    remaining_future.cancel()
            # 跳出循环，不再处理后续结果
            break

这个方案的核心是：用as_completed实时获取完成的任务结果，一旦达到目标长度，就取消所有未执行的任务，然后停止遍历。

方案2：终止正在运行的任务（需修改load_bucket） #

如果希望连正在运行的任务也能终止，你需要给load_bucket添加中断逻辑，比如用线程事件（threading.Event）来让任务主动检查是否需要停止：

from concurrent.futures import ThreadPoolExecutor, as_completed
import threading

# 修改load_bucket，加入停止事件检查
def load_bucket(bucket, stop_event):
    # 在任务开始时检查是否需要停止
    if stop_event.is_set():
        return None
    
    # 假设你的任务是循环处理内容，中间可以插入检查
    for item in bucket.get_items():  # 替换成你的实际处理逻辑
        if stop_event.is_set():
            # 主动终止任务
            break
        # 处理item的逻辑...
    
    # 返回处理结果
    return processed_result

res_list = []
target_length = number_reviews + bucket_size
# 创建一个线程事件，用于通知任务停止
stop_event = threading.Event()

with ThreadPoolExecutor() as pool:
    # 提交任务时传入stop_event
    futures = [pool.submit(load_bucket, bucket, stop_event) for bucket in buckets]
    
    for future in as_completed(futures):
        if len(res_list) < target_length:
            result = future.result()
            res_list.append(result)
        else:
            # 设置停止事件，通知所有任务终止
            stop_event.set()
            # 取消未完成的任务
            for remaining_future in futures:
                if not remaining_future.done():
                    remaining_future.cancel()
            break

为什么原来的map方法不行？ #

再啰嗦两句：pool.map()的设计是批量提交任务，然后按输入顺序返回结果。它会先把所有buckets对应的任务都塞进线程池，之后才开始逐个返回结果。所以你在循环里的break只是停止接收后续结果，但线程池里的任务已经全部启动了，根本没法中途停止任务提交或者终止任务。这就是为什么你的代码达不到预期效果的原因。

内容的提问来源于stack exchange，提问作者An old man in the sea.