Celery Rate Limiting: Controlling Task Execution Rates in Workers

In this tutorial, you will learn about Celery Rate Limiting: Controlling Task Execution Rates in Workers. We cover key concepts, practical examples, and best practices to help you master this topic.

Celery rate limiting controls how frequently tasks execute using per-task token bucket limits and per-worker rate limits, protecting downstream APIs and databases from being overwhelmed by burst task execution.

flowchart LR
    Queue[Task Queue] --> TokenBucket{Token Bucket}
    TokenBucket -->|Tokens Available| Execute[Execute Task]
    TokenBucket -->|No Tokens| Wait[Wait for Token]
    Wait --> TokenBucket
    Execute --> Consumer[API / Database]
    Consumer -->|Rate Limit Respected| Status[Stable System]

What You'll Learn

• Per-task rate limits with the rate_limit option
• Token bucket algorithm in Celery
• Per-worker rate limit configuration
• Application-level rate limiting strategies

Why It Matters

Without rate limiting, a burst of Celery tasks can overwhelm a downstream API or database. A task processing loop sending 1000 requests per second to an API limited to 10 req/s will get rate-limited or blocked, causing all tasks to fail. Rate limiting spreads execution to match the downstream capacity.

Real-World Use

DodaTech's email-sending Celery tasks use a rate limit of 10/m (10 per minute). This respects the email provider's API limit of 10 req/s. Without rate limiting, a bulk email campaign would trigger API rate limits and cause all email tasks to fail with 429 errors.

Per-Task Rate Limits

Set rate limits directly on task definitions:

from celery import Celery

app = Celery('ratelimit', broker='redis://localhost:6379/0')

@app.task(rate_limit='10/m')
def send_email(recipient, subject):
    import time
    time.sleep(0.5)
    print(f"Email sent to {recipient}: {subject}")
    return f"Sent to {recipient}"

@app.task(rate_limit='100/h')
def sync_external_api(data):
    import time
    time.sleep(0.2)
    print(f"Synced: {data[:50]}...")
    return "Synced"

@app.task(rate_limit='1/s')
def process_webhook(event):
    import time
    time.sleep(0.1)
    print(f"Webhook: {event['type']}")
    return "Processed"

# Simulate sending emails
import time as _time
for i in range(5):
    result = send_email.delay(f"user{i}@example.com", f"Message {i}")
    print(f"Queued: email to user{i}")
    _time.sleep(1)

Expected output:

Queued: email to user0
Email sent to user0@example.com: Message 0
Queued: email to user1
Email sent to user1@example.com: Message 1
... (rate limited to 10/min, so approximately one every 6 seconds)

Dynamic Rate Limits with Token Bucket

Override rate limits at call time:

from celery import Celery

app = Celery('ratelimit', broker='redis://localhost:6379/0')

@app.task
def call_external_api(endpoint, payload):
    import time
    time.sleep(0.1)
    print(f"API call to {endpoint}: {payload}")
    return {"status": "ok", "endpoint": endpoint}

# Override rate limit at call time based on endpoint
endpoint_limits = {
    "fast-api": "100/m",
    "slow-api": "5/m",
    "legacy-api": "1/m",
}

for endpoint, limit in endpoint_limits.items():
    task = call_external_api.s(endpoint, {"data": "test"})
    task.set(rate_limit=limit)
    task.delay()
    print(f"Queued {endpoint} with limit {limit}")

import time
time.sleep(0.5)

Expected output:

Queued fast-api with limit 100/m
Queued slow-api with limit 5/m
Queued legacy-api with limit 1/m
API call to fast-api: {'data': 'test'}
API call to slow-api: {'data': 'test'}
API call to legacy-api: {'data': 'test'}

Worker-Level Rate Limiting

Configure rate limits at the worker level:

from celery import Celery
import subprocess
import time

app = Celery('ratelimit', broker='redis://localhost:6379/0')

@app.task
def generic_task(task_id, data):
    import time
    time.sleep(0.05)
    print(f"Task {task_id} completed")
    return task_id

# Worker level rate limiting can be set with:
# celery -A ratelimit worker --rate-limit=generic_task=50/m

# Query current rate limit configuration
def get_worker_rate_limits():
    from celery.app.control import Inspect
    i = Inspect(app=app)
    stats = i.stats()
    if stats:
        for worker, info in stats.items():
            return info.get("rate_limits", {})
    return {}

for i in range(10):
    generic_task.delay(i, f"data_{i}")

time.sleep(0.5)

limits = get_worker_rate_limits()
print(f"Active rate limits: {limits}")

Expected output:

Task 0 completed
Task 1 completed
...
Task 9 completed
Active rate limits: {}

Common Mistakes

• Using rate limits that are too restrictive for task duration — if a task takes 30 seconds and the rate limit is 2/m, only 2 tasks can run per minute but each ties up a worker for 30 seconds. Match rate limits to task duration and worker count.
• Confusing 10/m with 10/s — 10/m means 10 tasks per minute (one every 6 seconds). 10/s means 10 per second. A typo here dramatically changes behavior.
• Not considering worker concurrency — with 4 workers and rate limit 10/m, each worker runs 10 tasks per minute, totaling 40/min. Rate limits are per-worker, not global.
• Setting rate limits on all tasks unnecessarily — most internal tasks do not need rate limits. Only rate-limit tasks that access external resources with capacity constraints.
• Forgetting that rate limits increase latency — if the queue grows faster than the rate limit, tasks wait in the queue. Monitor queue depth to detect rate limit bottlenecks.

Practice Questions

1. How does Celery rate limiting work at the task level?
2. What is the difference between per-task and per-worker rate limiting?
3. How does the token bucket algorithm smooth out task execution?
4. Why should rate limits match downstream capacity, not upstream demand?
5. How does worker concurrency affect effective rate limit enforcement?

Challenge

Design a rate limiting Strategy for a Celery-powered data pipeline that: (1) fetches data from 3 external APIs with different limits (API-A: 100/min, API-B: 10/min, API-C: 1/min), (2) transforms the data (no rate limit needed, CPU-bound), (3) writes to a database with a Connection Pool of 5, (4) sends email notifications with a limit of 20/min. Configure per-task rate limits, calculate minimum worker count to meet a throughput of 100 items/hour, and design a monitoring dashboard for rate limit metrics.

FAQ

Mini Project

Build a Celery application that interacts with 3 external APIs with different rate limits. Implement: (1) per-task rate limits matching each API's capacity, (2) a monitoring task that reports rate limit queue depth and wait times, (3) auto-throttling that reduces the rate limit when the API returns 429 responses, (4) a dynamic rate limit adjuster that increases limits during off-peak hours, and (5) a dashboard showing effective throughput vs configured rate limit for each task type.

What's Next

Continue with Celery Task Sets to learn about managing related task groups. Then explore Celery Events for the Celery event system.