如何为aiobotocore/boto3创建自定义重试逻辑解决S3 SlowDown错误？

核心思路 #

你的问题根源是S3请求频率超限触发SlowDown，官方默认重试策略未适配你的场景，逐个方法加重试逻辑确实冗余，以下是boto3和aiobotocore的全局自定义重试实现方案：

Boto3（同步）全局自定义重试 #

方法1：自定义RetryConfig规则 #

直接扩展botocore的重试配置，针对SlowDown错误设置专属退避策略：

import boto3
import random
from botocore.config import Config
from botocore.retries import standard

def custom_backoff(attempts):
    # 指数退避+随机抖动，最大延迟30秒
    return min(2 ** attempts + random.uniform(0, 1), 30)

# 针对SlowDown错误的自定义重试规则
slowdown_retry_rule = standard.RetryRule(
    exceptions=['SlowDown'],
    max_attempts=10,
    backoff=custom_backoff
)

# 合并默认规则与自定义规则
retry_config = standard.RetryConfig(
    mode='adaptive',
    rules=[slowdown_retry_rule] + standard.DEFAULT_RETRY_RULES
)

# 创建带全局重试配置的S3客户端
s3_config = Config(retry_config=retry_config)
s3_client = boto3.client('s3', config=s3_config)

方法2：通过事件系统注入重试逻辑 #

利用botocore的事件机制，全局拦截PutObject的重试判断：

import boto3
import time
from botocore.handlers import retry_handler

def custom_retry_handler(event, context):
    response = event['response']
    if response.get('error_code') == 'SlowDown':
        # 执行默认重试逻辑后，追加自定义延迟
        retry_handler(event, context)
        delay = min(2 ** context['attempt_number'], 30)
        time.sleep(delay + random.uniform(0, 1))

# 注册全局重试处理器
session = boto3.Session()
event_system = session.get_component('event_system')
event_system.register('needs-retry.s3.PutObject', custom_retry_handler)

s3_client = session.client('s3')

Aiobotocore（异步）全局自定义重试 #

方法1：配置自定义RetryConfig #

同步场景的配置逻辑可直接适配异步客户端：

import asyncio
import random
import aioboto3
from botocore.config import Config
from botocore.retries import standard

def custom_backoff(attempts):
    return min(2 ** attempts + random.uniform(0, 1), 30)

slowdown_retry_rule = standard.RetryRule(
    exceptions=['SlowDown'],
    max_attempts=10,
    backoff=custom_backoff
)

retry_config = standard.RetryConfig(
    mode='adaptive',
    rules=[slowdown_retry_rule] + standard.DEFAULT_RETRY_RULES
)

s3_config = Config(retry_config=retry_config)

async def upload_task():
    async with aioboto3.client('s3', config=s3_config) as client:
        await client.put_object(Bucket='your-bucket', Key='target-key', Body=b'file-content')

asyncio.run(upload_task())

方法2：全局异步重试装饰器 #

用tenacity给PutObject绑定全局重试逻辑：

import asyncio
import aioboto3
from tenacity import retry, stop_after_attempt, wait_exponential_jitter

# 定义全局重试规则
retry_decorator = retry(
    stop=stop_after_attempt(10),
    wait=wait_exponential_jitter(multiplier=1, max=30),
    retry=lambda state: (
        state.outcome.exception() is not None and
        getattr(state.outcome.exception(), 'response', {}).get('Error', {}).get('Code') == 'SlowDown'
    )
)

async def upload_task():
    async with aioboto3.client('s3') as client:
        # 给PutObject方法绑定重试装饰器
        client.put_object = retry_decorator(client.put_object)
        await client.put_object(Bucket='your-bucket', Key='target-key', Body=b'file-content')

asyncio.run(upload_task())

额外优化建议 #

• 限制并发数：同步场景用线程池控制并发量，异步场景用asyncio.Semaphore
• 分段上传：大文件改用create_multipart_upload拆分请求，降低单请求负载
• 动态限流：遇到SlowDown时临时降低并发数，恢复后再逐步提升

内容的提问来源于stack exchange，提问作者Philip Couling