答案：通过装饰器、队列和第三方库实现多线程重试机制。使用自定义retry装饰器或tenacity库为函数添加重试逻辑，结合threading和queue.Queue实现线程安全的任务调度，确保任务失败后可重试或记录，提升程序健壮性。
在使用Python多线程时，任务执行过程中可能会因为网络波动、资源竞争或外部服务不稳定等原因导致失败。为了提升程序的健壮性，需要为多线程任务设计重试机制和容错处理方案。下面介绍几种实用的方法。
1. 使用装饰器实现函数级重试
通过自定义重试装饰器，可以为可能失败的函数添加自动重试逻辑。结合异常捕获和延迟等待，提高任务成功率。
示例代码：
立即学习“Python免费学习笔记（深入）”；
import time
import random
from functools import wraps
<p>def retry(max_retries=3, delay=1, backoff=2, exceptions=(Exception,)):
def decorator(func):
@wraps(func)
def wrapper(*args, *<em>kwargs):
retries, current_delay = 0, delay
while retries < max_retries:
try:
return func(</em>args, *<em>kwargs)
except exceptions as e:
retries += 1
if retries == max_retries:
print(f"【失败】{func.<strong>name</strong>} 超出重试次数，错误: {e}")
raise e
print(f"【重试】{func.<strong>name</strong>} 第{retries}次，错误: {e}")
time.sleep(current_delay)
current_delay </em>= backoff # 指数退避
return None
return wrapper
return decorator</p><p>@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</p>2. 线程中调用带重试的任务函数
在线程中直接调用被@retry装饰的函数，每个线程独立处理自己的重试逻辑，互不干扰。
示例：
import threading
<p>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}")</p><h1>创建多个线程执行任务</h1><p>threads = []
for i in range(5):
t = threading.Thread(target=worker, args=(i+1,), name=f"Thread-{i+1}")
threads.append(t)
t.start()</p><p>for t in threads:
t.join()</p>3. 结合队列实现线程安全的容错任务调度
使用queue.Queue管理任务，配合线程池消费任务。任务失败可选择重新入队或记录日志，实现更灵活的容错控制。
优点：避免重复创建线程，支持动态任务分发，失败任务可重新调度。
import queue
import threading
import logging
<p>logging.basicConfig(level=logging.INFO)</p><p>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</p><pre class='brush:python;toolbar:false;'> try:
risky_task(task_id) # 带重试逻辑的函数
task_queue.task_done()
logging.info(f"任务 {task_id} 完成")
except Exception as e:
if retry_count < 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
<p>@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} 成功")</p>在多线程中调用该函数，tenacity会自动处理重试流程。
基本上就这些。核心是把重试逻辑封装好，无论是用装饰器、队列还是第三方库，关键是让每个线程具备独立的错误恢复能力，同时避免无限重试或资源浪费。
