Deleting all pending tasks in celery?

Celery Tasks

Task Management

Software Development

Python Programming

Debugging

Deleting all pending tasks in celery?

System Design practice on Codemia

Work through 120+ system design problems with detailed solutions, from rate limiters to multi-region storage.

Practice system design

Kafka & Streaming Python Debugging & Errors

Introduction

The fastest way to delete all pending Celery tasks is app.control.purge(), which removes every message from the broker queue. For finer control, you can revoke specific tasks by ID, purge individual named queues, or use the celery purge command-line tool. Each approach has different implications for running workers and active tasks, so choosing the right method depends on whether you need a clean slate or selective cancellation.

Celery Task Lifecycle

Before deleting tasks, it helps to understand where they live in the system:

State	Location	Can Be Deleted?
Pending (queued)	Broker (RabbitMQ, Redis, etc.)	Yes, via purge
Reserved (prefetched by worker)	Worker memory	Yes, via revoke
Active (currently executing)	Worker process	Yes, via revoke with terminate=True
Completed / Failed	Result backend	No (already finished)

Purging only removes messages from the broker queue. Tasks that a worker has already fetched (reserved) or started executing (active) require revocation.

Method 1: Purge All Queues Programmatically

The most common approach. This removes all pending messages from the default queue:

python

1from celery import Celery
2
3app = Celery('myapp', broker='redis://localhost:6379/0')
4
5# Purge all messages from the default queue
6discarded = app.control.purge()
7print(f"Purged {discarded} messages")

To target a specific queue:

python

1from kombu import Connection
2
3def purge_queue(broker_url, queue_name):
4    """Purge all messages from a specific queue."""
5    with Connection(broker_url) as conn:
6        simple_queue = conn.SimpleQueue(queue_name)
7        count = simple_queue.clear()
8        simple_queue.close()
9        print(f"Purged {count} messages from '{queue_name}'")
10
11purge_queue('redis://localhost:6379/0', 'celery')
12purge_queue('redis://localhost:6379/0', 'high_priority')

When using RabbitMQ, you can also purge through the management API:

bash

1# Purge the default 'celery' queue
2rabbitmqadmin purge queue name=celery
3
4# List all queues and their message counts
5rabbitmqadmin list queues name messages

Method 2: Celery Command-Line Tool

The celery purge command is the simplest option when you have shell access:

bash

1# Purge the default queue (prompts for confirmation)
2celery -A myapp purge
3
4# Force purge without confirmation
5celery -A myapp purge -f
6
7# Purge specific queues
8celery -A myapp purge -Q high_priority,low_priority

To inspect what is in the queue before purging:

bash

1# Show active, reserved, and scheduled tasks
2celery -A myapp inspect active
3celery -A myapp inspect reserved
4celery -A myapp inspect scheduled
5
6# Show queue lengths (requires the inspect command)
7celery -A myapp inspect stats

Method 3: Revoke Specific Tasks

When you need to cancel specific tasks rather than purging everything, use revoke with the task ID:

python

1from celery import Celery
2
3app = Celery('myapp', broker='redis://localhost:6379/0')
4
5# Revoke a pending task (removes from queue)
6app.control.revoke('task-id-123')
7
8# Revoke and terminate an active task
9app.control.revoke('task-id-456', terminate=True)
10
11# Revoke and terminate with SIGKILL (last resort)
12app.control.revoke('task-id-789', terminate=True, signal='SIGKILL')

To revoke multiple tasks at once:

python

# Revoke a batch of tasks
task_ids = ['id-1', 'id-2', 'id-3', 'id-4']
app.control.revoke(task_ids)

For AsyncResult objects returned when you dispatch tasks:

python

result = my_task.delay(arg1, arg2)
# Later, cancel it
result.revoke(terminate=True)

Method 4: Revoke by Task Name

To revoke all instances of a specific task type, combine inspection with revocation:

python

1def revoke_all_by_name(app, task_name):
2    """Revoke all pending and active instances of a named task."""
3    inspector = app.control.inspect()
4    
5    revoked = 0
6    
7    # Check reserved (prefetched) tasks
8    reserved = inspector.reserved() or {}
9    for worker, tasks in reserved.items():
10        for task in tasks:
11            if task['name'] == task_name:
12                app.control.revoke(task['id'], terminate=True)
13                revoked += 1
14    
15    # Check active (running) tasks
16    active = inspector.active() or {}
17    for worker, tasks in active.items():
18        for task in tasks:
19            if task['name'] == task_name:
20                app.control.revoke(task['id'], terminate=True)
21                revoked += 1
22    
23    print(f"Revoked {revoked} instances of '{task_name}'")
24
25revoke_all_by_name(app, 'myapp.tasks.send_email')

Method 5: Flush the Broker Directly

When Celery commands are unavailable (worker is down, app cannot connect), you can flush the broker directly:

For Redis:

bash

1# Delete the default Celery queue key
2redis-cli DEL celery
3
4# Delete a named queue
5redis-cli DEL high_priority
6
7# Flush entire Redis database (nuclear option)
8redis-cli FLUSHDB

python

1import redis
2
3r = redis.Redis(host='localhost', port=6379, db=0)
4deleted = r.delete('celery')
5print(f"Deleted queue key: {deleted}")

For RabbitMQ:

bash

1# Purge via rabbitmqctl
2rabbitmqctl purge_queue celery
3
4# Delete the queue entirely
5rabbitmqctl delete_queue celery

Preventing Task Buildup

Rather than purging reactively, these patterns prevent excessive task accumulation:

Task Expiration

Set an expiration time so stale tasks are automatically discarded:

python

1# Per-task expiration
2@app.task(expires=3600)  # Expire after 1 hour
3def send_notification(user_id, message):
4    pass
5
6# Per-call expiration
7send_notification.apply_async(
8    args=[user_id, message],
9    expires=600  # Expire after 10 minutes
10)

Rate Limiting

Prevent task producers from overwhelming the queue:

python

@app.task(rate_limit='100/m')  # Max 100 executions per minute
def process_webhook(payload):
    pass

Task Deduplication

Avoid duplicate tasks using a lock:

python

1from celery import Celery
2import redis
3
4app = Celery('myapp', broker='redis://localhost:6379/0')
5redis_client = redis.Redis()
6
7@app.task(bind=True)
8def deduplicated_task(self, key, *args):
9    lock_key = f"task_lock:{key}"
10    if not redis_client.set(lock_key, self.request.id, nx=True, ex=3600):
11        print(f"Task for {key} already queued, skipping")
12        return
13    try:
14        # Do actual work
15        pass
16    finally:
17        redis_client.delete(lock_key)

Comparison of Methods

Method	Scope	Requires Workers	Safe for Production
`app.control.purge()`	All queued tasks	No	Use with caution
`celery purge` CLI	All or specific queues	No	Use with caution
`app.control.revoke(id)`	Single task	Yes (for active tasks)	Yes
Broker flush (Redis/RabbitMQ)	All data in broker	No	Last resort only
Task expiration	Stale tasks	No	Yes (preventive)

Common Pitfalls

Purging does not stop active tasks. app.control.purge() only removes messages from the broker queue. Tasks already being executed by workers continue running. Use revoke(terminate=True) for active tasks.
Worker prefetch hides tasks from purge. Workers prefetch tasks from the broker into memory. Purging the queue does not affect these prefetched tasks. Either revoke them individually or restart the workers after purging.
Using terminate=True with SIGTERM may not work. If the task ignores SIGTERM (common with subprocess calls or C extensions), escalate to signal='SIGKILL'. Be aware that SIGKILL prevents cleanup code from running.
Confusing countdown with expires. countdown=120 delays execution by 120 seconds. expires=120 discards the task if not started within 120 seconds. They serve different purposes and are often confused.
Flushing Redis FLUSHDB deletes everything. If your Redis instance stores more than Celery data (cache, sessions, rate limits), FLUSHDB destroys all of it. Target specific queue keys instead.
Not monitoring queue depth. Set up alerts for queue length thresholds (e.g., Prometheus + Flower). Discovering thousands of stuck tasks retroactively is much harder than preventing buildup.

Summary

Use app.control.purge() or celery purge to remove all pending tasks from the broker queue.
Use app.control.revoke(task_id, terminate=True) to cancel specific tasks, including active ones.
Purging does not affect tasks already being executed or prefetched by workers. Restart workers after purging for a complete reset.
For broker-level operations, use redis-cli DEL or rabbitmqctl purge_queue when Celery commands are unavailable.
Prevent task buildup proactively with expires, rate limiting, and deduplication patterns.
Always inspect queue state with celery inspect before and after purging to verify the result.

System Design Fundamentals

Build a strong foundation in designing scalable, reliable distributed systems.

View the course

Track what you have practised

A free account saves your progress, solutions and study plan across every problem on Codemia.

System Design practice on Codemia

Work through 120+ system design problems with detailed solutions, from rate limiters to multi-region storage.

Practice system design

Deleting all pending tasks in celery?

System Design practice on Codemia

Introduction

Celery Task Lifecycle

Method 1: Purge All Queues Programmatically

Method 2: Celery Command-Line Tool

Method 3: Revoke Specific Tasks

Method 4: Revoke by Task Name

Method 5: Flush the Broker Directly

Preventing Task Buildup

Task Expiration

Rate Limiting

Task Deduplication

Comparison of Methods

Common Pitfalls

Summary

Related reading

System Design Fundamentals

Track what you have practised

System Design practice on Codemia