如何在Python的ThreadPoolExecutor中实现即时重试逻辑？

需要在Python的concurrent.futures.ThreadPoolExecutor中实现满足以下要求的重试逻辑：

• 任务失败后立即将新的重试任务加入工作队列
• 重试任务支持再次重试，可配置无限重试或最大重试次数

现有方案的问题 #

网上常见的轮询模式（如下代码）无法满足即时重试的要求——它需要等待当前批次所有任务完成后，才会处理重试任务。比如线程池有2个worker时，一个1秒完成且高失败率的任务，在第二次失败后，必须等另一个100秒的任务完成才能继续重试，完全浪费了空闲的线程资源。

with concurrent.futures.ThreadPoolExecutor(...) as executor:
    futures = {executor.submit(fn, job): job for job in jobs}
    while len(futures) > 0:
        new_futures = {}
        for fut in concurrent.futures.as_completed(futures):
            if fut.exception():
                job = futures[fut]
                new_futures[executor.submit(fn, job)] = job
            else:
                # 处理任务成功的逻辑
                pass
        futures = new_futures

你之前尝试的写法无效的原因 #

你之前在任务内部提交新任务的写法不生效，主要是两个问题：一是executor在with块内的作用域无法被内部任务正确引用，二是没做重试次数控制，同时map方法会等待所有初始任务完成后才返回，无法追踪后续提交的重试任务。

可行实现方案 #

可以通过在任务内部封装重试逻辑+传递执行器引用+控制重试次数的方式实现，核心思路是让失败的任务自己提交重试任务到线程池，同时记录已重试次数，达到最大次数后停止。

基础实现代码 #

import concurrent.futures
import functools

def retryable_task(executor, fn, max_retries=None, current_retry=0):
    try:
        # 执行原始任务
        result = fn()
        print(f"任务成功: {result}")
        return result
    except Exception as e:
        print(f"任务失败，重试次数 {current_retry}: {e}")
        # 判断是否继续重试
        if max_retries is None or current_retry < max_retries:
            # 立即提交重试任务到线程池队列
            executor.submit(
                retryable_task,
                executor,
                fn,
                max_retries,
                current_retry + 1
            )
        else:
            print(f"任务达到最大重试次数 {max_retries}，停止重试")
            # 处理最终失败的逻辑
            return None

# 示例：模拟90%失败率的任务
def sample_job(job_id):
    import random
    if random.random() < 0.9:
        raise ValueError(f"任务 {job_id} 执行失败")
    return f"任务 {job_id} 执行成功"

if __name__ == "__main__":
    jobs = [1, 2]
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
        # 提交初始任务
        for job_id in jobs:
            # 用partial绑定任务参数，生成可无参调用的函数
            task_fn = functools.partial(sample_job, job_id)
            executor.submit(
                retryable_task,
                executor,
                task_fn,
                max_retries=5  # 设置最大重试次数，传None则无限重试
            )
        # 等待所有任务（包括重试）执行完成
        executor.shutdown(wait=True)

关键细节 #

1. 即时重试：任务失败后立刻调用executor.submit把重试任务加入队列，只要有空闲worker就会马上执行，不用等当前批次任务全部完成。
2. 重试次数控制：通过max_retries参数设置最大重试次数，current_retry记录当前重试次数，每次重试自动递增。
3. 作用域问题解决：把executor作为参数传递给重试任务，避免内部引用失效的问题。
4. 参数绑定：用functools.partial把原始任务的参数绑定好，生成可以直接调用的函数，方便重试逻辑复用。

进阶：追踪所有任务结果 #

如果需要实时处理所有任务（包括重试任务）的结果，可以维护一个队列来记录所有提交的future，然后用as_completed实时处理：

import concurrent.futures
import functools
from collections import deque

def retryable_task(executor, fn, max_retries=None, current_retry=0, future_queue=None):
    try:
        result = fn()
        print(f"任务成功: {result}")
        return result
    except Exception as e:
        print(f"任务失败，重试次数 {current_retry}: {e}")
        if max_retries is None or current_retry < max_retries:
            new_fut = executor.submit(
                retryable_task,
                executor,
                fn,
                max_retries,
                current_retry + 1,
                future_queue
            )
            if future_queue is not None:
                future_queue.append(new_fut)
        else:
            print(f"任务达到最大重试次数 {max_retries}，停止重试")
            return None

if __name__ == "__main__":
    jobs = [1, 2]
    future_queue = deque()

    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
        for job_id in jobs:
            task_fn = functools.partial(sample_job, job_id)
            fut = executor.submit(
                retryable_task,
                executor,
                task_fn,
                max_retries=5,
                future_queue=future_queue
            )
            future_queue.append(fut)
        
        # 实时处理所有任务结果
        while future_queue:
            fut = next(concurrent.futures.as_completed(future_queue))
            future_queue.remove(fut)
            # 在这里处理任务结果，比如记录到数据库、统计等
            result = fut.result()

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