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>

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>

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() # 等待所有任务完成

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>