IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Python脚本如何重试分布式同步失败任务

wen python案例 215

本文目录导读：

Python脚本如何重试分布式同步失败任务

  1. 基础重试机制 - 使用tenacity库
  2. 带任务队列的分布式重试
  3. 使用消息队列实现（RabbitMQ示例）
  4. 高级重试策略 - 自定义装饰器
  5. 完整示例 - 生产级实现
  6. 建议

对于分布式同步失败任务的重试,有几种常见且实用的方案，我来详细介绍几种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

建议

  1. 选择合适的重试策略：

    • 网络波动：使用指数退避
    • 服务临时不可用：使用固定间隔
    • 数据一致性问题：使用立即重试+最终重试

  2. 监控和告警：

    • 记录重试次数和原因
    • 设置死信队列告警
    • 监控重试队列长度

  3. 幂等性设计：

    • 确保重试操作是幂等的
    • 使用唯一ID防止重复处理

根据你的具体场景选择合适的方案,生产环境建议使用消息队列+Redis组合方案。

版权声明：NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 NBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 CBA直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播

上一篇Python脚本如何实现分布式同步容错机制

下一篇Python脚本如何恢复节点后继续同步任务

相关文章

抱歉，评论功能暂时关闭!

友情链接攻略窝 · 礼包窝 · 配置窝 · 换算窝 · 算法窝 · 查文窝 · 手册窝 · 问答窝 · 全文搜 · 手册库
相关站点礼包窝 · 查文窝