本文目录导读：
对于分布式同步失败任务的重试,有几种常见且实用的方案，我来详细介绍几种Python实现方式：
基础重试机制 - 使用tenacity库
import time
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
class DistributedSyncError(Exception):
"""分布式同步异常"""
pass
class SyncWorker:
def __init__(self, max_retries=3):
self.max_retries = max_retries
self.task_queue = []
@retry(
stop=stop_after_attempt(3), # 最多重试3次
wait=wait_exponential(multiplier=1, min=1, max=10), # 指数退避
retry=retry_if_exception_type((DistributedSyncError, ConnectionError))
)
def sync_task(self, task_data):
"""执行同步任务，失败自动重试"""
try:
# 模拟分布式同步
result = self._do_sync(task_data)
return result
except Exception as e:
print(f"同步失败: {e}")
raise DistributedSyncError(f"同步任务失败: {e}")
def _do_sync(self, task_data):
# 模拟同步逻辑
import random
if random.random() < 0.7: # 70%概率失败
raise ConnectionError("网络连接失败")
return {"status": "success", "data": task_data}
带任务队列的分布式重试
import redis
import json
import pickle
from datetime import datetime, timedelta
import threading
class DistributedRetryManager:
def __init__(self, redis_host='localhost', redis_port=6379):
self.redis_client = redis.StrictRedis(
host=redis_host,
port=redis_port,
decode_responses=True
)
self.retry_queue_key = "sync:retry:queue"
self.dead_letter_queue = "sync:dead:letter"
def enqueue_task(self, task_data, max_retries=3):
"""将任务加入重试队列"""
task = {
'data': task_data,
'retry_count': 0,
'max_retries': max_retries,
'timestamp': datetime.now().isoformat(),
'next_retry_time': datetime.now().isoformat()
}
self.redis_client.rpush(self.retry_queue_key, json.dumps(task))
def process_retry_queue(self, worker_func):
"""处理重试队列"""
while True:
# 获取待重试的任务
task_json = self.redis_client.lpop(self.retry_queue_key)
if not task_json:
time.sleep(5)
continue
task = json.loads(task_json)
try:
# 检查重试时间
next_retry = datetime.fromisoformat(task['next_retry_time'])
if datetime.now() < next_retry:
# 还没到重试时间，放回队列
self.redis_client.rpush(self.retry_queue_key, json.dumps(task))
continue
# 执行任务
result = worker_func(task['data'])
print(f"任务执行成功: {task['data']}")
except Exception as e:
task['retry_count'] += 1
if task['retry_count'] < task['max_retries']:
# 计算下次重试时间（指数退避）
wait_seconds = 2 ** task['retry_count'] * 10
next_retry_time = datetime.now() + timedelta(seconds=wait_seconds)
task['next_retry_time'] = next_retry_time.isoformat()
task['last_error'] = str(e)
# 重新放入队列
self.redis_client.rpush(self.retry_queue_key, json.dumps(task))
print(f"任务重试 {task['retry_count']}/{task['max_retries']}, 等待{wait_seconds}秒")
else:
# 超过最大重试次数，放入死信队列
self.redis_client.lpush(self.dead_letter_queue, json.dumps(task))
print(f"任务移入死信队列: {task['data']}")
使用消息队列实现（RabbitMQ示例）
import pika
import json
from functools import wraps
class RabbitMQRetryHandler:
def __init__(self, host='localhost'):
self.connection = pika.BlockingConnection(
pika.ConnectionParameters(host=host)
)
self.channel = self.connection.channel()
# 主队列
self.channel.queue_declare(queue='sync_tasks', durable=True)
# 死信队列
self.channel.queue_declare(queue='sync_tasks_retry', durable=True)
def publish_task(self, task_data, retry_count=0):
"""发布任务"""
message = {
'data': task_data,
'retry_count': retry_count
}
self.channel.basic_publish(
exchange='',
routing_key='sync_tasks',
body=json.dumps(message),
properties=pika.BasicProperties(
delivery_mode=2, # 持久化消息
headers={'retry_count': retry_count}
)
)
def process_messages(self, worker_func):
"""处理消息"""
def callback(ch, method, properties, body):
task = json.loads(body)
try:
result = worker_func(task['data'])
ch.basic_ack(delivery_tag=method.delivery_tag)
print(f"任务处理成功: {task['data']}")
except Exception as e:
retry_count = task.get('retry_count', 0) + 1
if retry_count <= 3: # 最多重试3次
# 重新发布到队列
self.publish_task(task['data'], retry_count)
print(f"任务重试 {retry_count}/3")
else:
# 移入死信队列
self.channel.basic_publish(
exchange='',
routing_key='sync_tasks_retry',
body=body,
properties=pika.BasicProperties(
delivery_mode=2
)
)
print(f"任务移入死信队列: {task['data']}")
ch.basic_ack(delivery_tag=method.delivery_tag)
self.channel.basic_consume(
queue='sync_tasks',
on_message_callback=callback
)
print('等待任务...')
self.channel.start_consuming()
高级重试策略 - 自定义装饰器
import functools
import logging
import random
from datetime import datetime
logger = logging.getLogger(__name__)
def distributed_retry(
max_retries=3,
base_delay=1.0,
max_delay=60.0,
backoff_factor=2,
jitter=True,
retryable_exceptions=(Exception,)
):
"""
分布式系统重试装饰器
参数:
- max_retries: 最大重试次数
- base_delay: 基础延迟
- max_delay: 最大延迟
- backoff_factor: 退避因子
- jitter: 是否添加随机抖动
- retryable_exceptions: 可重试的异常类型
"""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_retries + 1):
try:
return func(*args, **kwargs)
except retryable_exceptions as e:
last_exception = e
if attempt == max_retries:
logger.error(f"任务失败，已达最大重试次数 {max_retries}")
raise
# 计算延迟时间
delay = min(
base_delay * (backoff_factor ** attempt),
max_delay
)
# 添加随机抖动，避免惊群效应
if jitter:
delay *= (0.5 + random.random())
logger.warning(
f"任务失败，尝试 {attempt + 1}/{max_retries}, "
f"等待 {delay:.2f} 秒, 错误: {e}"
)
time.sleep(delay)
raise last_exception
return wrapper
return decorator
# 使用示例
@distributed_retry(
max_retries=5,
base_delay=1,
max_delay=30,
backoff_factor=2,
jitter=True,
retryable_exceptions=(ConnectionError, TimeoutError)
)
def sync_distributed_data(data):
"""同步分布式数据"""
# 模拟分布式操作
if random.random() < 0.8: # 80%概率失败
raise ConnectionError("网络连接失败")
return f"数据 {data} 同步成功"
完整示例 - 生产级实现
import asyncio
import aiohttp
import logging
from typing import Callable, Any, Optional
from dataclasses import dataclass
from enum import Enum
class SyncStatus(Enum):
PENDING = "pending"
SUCCESS = "success"
FAILED = "failed"
RETRYING = "retrying"
@dataclass
class SyncTask:
id: str
data: Any
status: SyncStatus = SyncStatus.PENDING
retry_count: int = 0
max_retries: int = 3
last_error: Optional[str] = None
class DistributedSyncManager:
def __init__(self, task_store=None):
self.task_store = task_store or {}
self.logger = logging.getLogger(__name__)
async def execute_with_retry(
self,
task: SyncTask,
sync_func: Callable
) -> bool:
"""异步执行同步任务并重试"""
while task.retry_count <= task.max_retries:
try:
task.status = SyncStatus.RETRYING
result = await sync_func(task.data)
if result:
task.status = SyncStatus.SUCCESS
self.logger.info(f"任务 {task.id} 同步成功")
return True
except aiohttp.ClientError as e:
task.retry_count += 1
task.last_error = str(e)
if task.retry_count <= task.max_retries:
wait_time = min(30, 2 ** task.retry_count)
self.logger.warning(
f"任务 {task.id} 第 {task.retry_count} 次重试, "
f"等待 {wait_time} 秒"
)
await asyncio.sleep(wait_time)
else:
task.status = SyncStatus.FAILED
self.logger.error(f"任务 {task.id} 最终失败: {e}")
return False
return False
async def sync_batch(self, tasks: list, sync_func: Callable):
"""批量同步任务"""
tasks_to_retry = []
for task in tasks:
success = await self.execute_with_retry(task, sync_func)
if not success:
tasks_to_retry.append(task)
# 处理需要重试的任务
if tasks_to_retry:
self.logger.info(f"需要重试 {len(tasks_to_retry)} 个任务")
await self.retry_failed_tasks(tasks_to_retry, sync_func)
async def retry_failed_tasks(self, tasks: list, sync_func: Callable):
"""重试失败的任务"""
while tasks:
await asyncio.sleep(60) # 等待1分钟后重试
remaining_tasks = []
for task in tasks:
if task.retry_count < task.max_retries:
success = await self.execute_with_retry(task, sync_func)
if not success:
remaining_tasks.append(task)
else:
self.logger.error(f"任务 {task.id} 已放弃重试")
tasks = remaining_tasks
建议
-
选择合适的重试策略：
- 网络波动：使用指数退避
- 服务临时不可用：使用固定间隔
- 数据一致性问题：使用立即重试+最终重试
-
监控和告警：
- 记录重试次数和原因
- 设置死信队列告警
- 监控重试队列长度
-
幂等性设计：
- 确保重试操作是幂等的
- 使用唯一ID防止重复处理
根据你的具体场景选择合适的方案,生产环境建议使用消息队列+Redis组合方案。
