The problem
You need to retry distributed operations while preventing thundering herd problem.
The solution
import functools
import random
import time
from typing import Tuple, Type
def retry_with_jitter(
max_attempts: int = 3,
base_delay: float = 1.0,
max_delay: float = 60.0,
backoff: float = 2.0,
jitter: float = 0.1,
exceptions: Tuple[Type[Exception], ...] = (Exception,)
):
"""Retry decorator with exponential backoff and jitter."""
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
last_exception = None
for attempt in range(max_attempts):
try:
return func(*args, **kwargs)
except exceptions as e:
last_exception = e
if attempt == max_attempts - 1:
break
# Calculate backoff with jitter
delay = min(
base_delay * (backoff ** attempt),
max_delay
)
# Add jitter (±jitter%)
jitter_amount = delay * jitter
delay_with_jitter = delay + random.uniform(-jitter_amount, jitter_amount)
delay_with_jitter = max(0, delay_with_jitter)
time.sleep(delay_with_jitter)
raise last_exception
return wrapper
return decoratorParameters
max_attemptsintMaximum number of retry attempts
base_delayfloatInitial delay in seconds
max_delayfloatMaximum delay in seconds
backofffloatExponential backoff multiplier
jitterfloatJitter factor (e.g., 0.1 for ±10%)
exceptionstupleException types to retry
Put it to work
import requests
@retry_with_jitter(
max_attempts=5,
base_delay=1.0,
max_delay=30.0,
jitter=0.2,
exceptions=(requests.RequestException,)
)
def fetch_with_jitter(url):
response = requests.get(url, timeout=10)
response.raise_for_status()
return response.json()
# Will retry with random delays to prevent thundering herd
data = fetch_with_jitter('https://api.example.com/data')