答案：通过装饰器、队列和第三方库实现多线程重试机制。使用自定义retry装饰器或tenacity库为函数添加重试逻辑，结合threading和queue.Queue实现线程安全的任务调度，确保任务失败后可重试或记录，提升程序健壮性。
在使用Python多线程时，任务执行过程中可能会因为网络波动、资源竞争或外部服务不稳定等原因导致失败。为了提升程序的健壮性，需要为多线程任务设计重试机制和容错处理方案。下面介绍几种实用的方法。
1. 使用装饰器实现函数级重试
通过自定义重试装饰器，可以为可能失败的函数添加自动重试逻辑。结合异常捕获和延迟等待，提高任务成功率。
示例代码：
import time import random from functools import wraps
def retry(max_retries=3, delay=1, backoff=2, exceptions=(Exception,)): def decorator(func): @wraps(func) def wrapper(*args, *kwargs): retries, current_delay = 0, delay while retries < max_retries: try: return func(args, *kwargs) except exceptions as e: retries += 1 if retries == max_retries: print(f"【失败】{func.name} 超出重试次数，错误: {e}") raise e print(f"【重试】{func.name} 第{retries}次，错误: {e}") time.sleep(current_delay) current_delay = backoff # 指数退避 return None return wrapper return decorator
@retry(max_retries=3, delay=1) def risky_task(task_id): if random.random() < 0.7: # 70%概率失败 raise ConnectionError("模拟网络错误") print(f"✅ 任务 {task_id} 成功完成") return task_id
e
import random
from functools import wraps
2. 线程中调用带重试的任务函数
在线程中直接调用被@retry装饰的函数，每个线程独立处理自己的重试逻辑，互不干扰。
示例：
import threadingdef worker(task_id): try: result = risky_task(task_id) print(f"? 线程 {threading.current_thread().name} 完成任务 {result}") except Exception as e: print(f"❌ 任务 {task_id} 最终失败: {e}")
创建多个线程执行任务
threads = [] for i in range(5): t = threading.Thread(target=worker, args=(i+1,), name=f"Thread-{i+1}") threads.append(t) t.start()
for t in threads: t.join()
3. 结合队列实现线程安全的容错任务调度
使用queue.Queue管理任务，配合线程池消费任务。任务失败可选择重新入队或记录日志，实现更灵活的容错控制。
优点：避免重复创建线程，支持动态任务分发，失败任务可重新调度。
import queue import threading import logginglogging.basicConfig(level=logging.INFO)
def worker_with_queue(task_queue, max_retries=3): while True: try: task_id, retry_count = task_queue.get(timeout=2) except queue.Empty: break
try: risky_task(task_id) # 带重试逻辑的函数 task_queue.task_done() logging.info(f"任务 {task_id} 完成") except Exception as e: if retry_count zuojiankuohaophpcn max_retries: new_count = retry_count + 1 task_queue.put((task_id, new_count)) logging.warning(f"任务 {task_id} 将重试第 {new_count} 次") else: logging.error(f"任务 {task_id} 彻底失败") task_queue.task_done()初始化任务队列
q = queue.Queue() for i in range(3): q.put((i+1, 0)) # (任务ID, 重试次数)
启动工作线程
for _ in range(2): t = threading.Thread(target=worker_with_queue, args=(q, 3), daemon=True) t.start()
q.join() # 等待所有任务完成
4. 使用第三方库简化重试逻辑
推荐使用 tenacity 库，功能强大且支持多线程环境。
安装：
pip install tenacity
使用示例：
from tenacity import retry, stop_after_attempt, wait_exponential@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, max=10)) def tenacity_task(task_id): if random.random() < 0.6: raise RuntimeError("模拟失败") print(f"✅ tenacity 任务 {task_id} 成功")
在多线程中调用该函数，tenacity会自动处理重试流程。
基本上就这些。核心是把重试逻辑封装好，无论是用装饰器、队列还是第三方库，关键是让每个线程具备独立的错误恢复能力，同时避免无限重试或资源浪费。
