How to Implement Retry with Exponential Backoff and Jitter in Python
This code demonstrates a retry mechanism with exponential backoff and optional full jitter, using a flaky mock network call for testing.
Python code
32 linesimport random
import time
def retry_with_backoff(func, max_attempts=5, base_delay=0.1, jitter=True):
"""
Retry a function with exponential backoff and optional full jitter.
"""
for attempt in range(max_attempts):
try:
return func()
except Exception as e:
if attempt == max_attempts - 1:
raise
delay = base_delay * (2 ** attempt)
if jitter:
delay = random.uniform(0, delay)
print(f"Attempt {attempt + 1} failed: {e}. Retrying in {delay:.4f}s")
time.sleep(delay)
def flaky_network_call():
"""Mock a network call that fails 60% of the time."""
if random.random() < 0.6:
raise ConnectionError("Network timeout")
return "Success"
if __name__ == "__main__":
random.seed(42) # For reproducible output
result = retry_with_backoff(flaky_network_call)
print(f"Final result: {result}")
Output
Attempt 1 failed: Network timeout. Retrying in 0.0338s
Attempt 2 failed: Network timeout. Retrying in 0.1718s
Final result: Success
How it works
The retry_with_backoff function iterates up to max_attempts, calling func() each time. On failure, it calculates an exponential delay based on the attempt number, then optionally applies full jitter by picking a random value between 0 and the base delay. This reduces thundering herd problems in distributed systems by preventing simultaneous retries. The flaky_network_call mock simulates a 60% failure rate, and with a fixed seed, the output becomes reproducible for testing.
Common mistakes
- Forgetting to re-raise the last exception after exhausting attempts
- Applying jitter incorrectly, e.g., adding it to the delay instead of randomizing within the full range
- Using `time.sleep` without understanding it blocks the thread, which can be problematic in async contexts
Variations
- Use `random.uniform(0, delay)` for full jitter vs. partial jitter `delay/2 + random.uniform(0, delay/2)`
- Replace `time.sleep` with `await asyncio.sleep` for async compatibility
Real-world use cases
- Retrying API calls to third-party services that experience transient failures or rate limits.
- Handling database connection timeouts in a microservice to avoid immediate cascading failures.
- Scheduling background job retries in a message queue consumer to smooth out load spikes.
Sponsored
More from System design patterns
- Build a BFF (Backend for Frontend) Mock Aggregator in Python medium
- Builder pattern for mocking complex objects in Python easy
- Circuit Breaker Pattern in Python: Closed, Open, and Half-Open States medium
- Create a Data Helper Class in Python easy
- Domain Driven Design Aggregate Root Example in Python medium
- Facade Pattern in Python with Mock Simplification medium
Keep learning
Related tutorials and quizzes for this topic.