答案：通过装饰器、队列和第三方库实现多线程重试机制。使用自定义retry装饰器或tenacity库为函数添加重试逻辑，结合threading和queue.Queue实现线程安全的任务调度，确保任务失败后可重试或记录，提升程序健壮性。

在使用Python多线程时，任务执行过程中可能会因为网络波动、资源竞争或外部服务不稳定等原因导致失败。为了提升程序的健壮性，需要为多线程任务设计重试机制和容错处理方案。下面介绍几种实用的方法。

1. 使用装饰器实现函数级重试

通过自定义重试装饰器，可以为可能失败的函数添加自动重试逻辑。结合异常捕获和延迟等待，提高任务成功率。

示例代码：

立即学习“Python免费学习笔记（深入）”；

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

2. 线程中调用带重试的任务函数

在线程中直接调用被@retry装饰的函数，每个线程独立处理自己的重试逻辑，互不干扰。

示例：

import threading
def 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管理任务，配合线程池消费任务。任务失败可选择重新入队或记录日志，实现更灵活的容错控制。

优点：避免重复创建线程，支持动态任务分发，失败任务可重新调度。

AI发型设计

虚拟发型试穿工具和发型模拟器

下载

import queue
import threading
import logging
logging.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会自动处理重试流程。
基本上就这些。核心是把重试逻辑封装好，无论是用装饰器、队列还是第三方库，关键是让每个线程具备独立的错误恢复能力，同时避免无限重试或资源浪费。