Celery Broadcast Messages: Sending Commands to All Workers in a Cluster

In this tutorial, you will learn about Celery Broadcast Messages: Sending Commands to All Workers in a Cluster. We cover key concepts, practical examples, and best practices to help you master this topic.

Celery broadcast enables sending commands to every worker in the cluster simultaneously, supporting custom broadcast handlers, targeted delivery to specific workers, and cluster-wide coordination for deployments, cache invalidation, and configuration updates.

flowchart TD
    S[Broadcast Sender] -->|Control Exchange| B[RabbitMQ/Redis]
    B -->|Fanout| W1[Worker 1]
    B -->|Fanout| W2[Worker 2]
    B -->|Fanout| W3[Worker 3]
    W1 -->|Reply| S
    W2 -->|Reply| S
    W3 -->|Reply| S

What You'll Learn

• Custom broadcast handlers
• Targeting specific workers
• Broadcast reply collection
• Use cases: cache invalidation, config reload
• Broadcast vs direct task patterns

Why It Matters

Without broadcast, sending a command to every worker requires submitting a task per worker with no guarantee of delivery. Broadcast provides reliable fanout delivery through the broker, with optional replies from each worker.

Real-World Use

DodaTech's platform sends a broadcast command to all workers when malware signatures update, triggering each worker to reload its signature cache from the shared database without requiring individual task submissions.

Custom Broadcast Handler

from celery import Celery
from celery.worker.control import Panel
import time

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

@Panel.register
def reload_cache(panel, **kwargs):
    print(f"Cache reload triggered on {panel.hostname}")
    return {'ok': f'cache_reloaded on {panel.hostname}'}

@Panel.register
def ping(panel, **kwargs):
    return {'pong': panel.hostname, 'timestamp': time.time()}

@app.task
def process_data(data_id):
    result = f"Processing {data_id}"
    print(result)
    return result

print("Registered custom broadcast handlers: reload_cache, ping")

Send broadcast from CLI:

celery -A broadcast control reload_cache
celery -A broadcast control ping

Expected output:

celery@host1: {'ok': 'cache_reloaded on celery@host1'}
celery@host2: {'ok': 'cache_reloaded on celery@host2'}
celery@host1: {'pong': 'celery@host1', 'timestamp': 1719561600.0}
celery@host2: {'pong': 'celery@host2', 'timestamp': 1719561600.1}

Broadcast via Control API

from celery import Celery
import json

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

def invalidate_cache_all(pattern='*'):
    result = app.control.broadcast(
        'reload_cache',
        arguments={'pattern': pattern},
        reply=True,
        timeout=10,
    )
    print(f"Cache invalidation sent for pattern '{pattern}'")
    for worker, reply in result.items():
        print(f"  {worker}: {reply['ok']}")
    return result

def ping_all_workers():
    result = app.control.broadcast(
        'ping',
        reply=True,
        timeout=5,
    )
    print(f"Pinged {len(result)} workers:")
    for worker, reply in result.items():
        print(f"  {reply['pong']} - {reply['timestamp']}")
    return result

invalidate_cache_all(pattern='user:*')
ping_all_workers()

Expected output:

Cache invalidation sent for pattern 'user:*'
  celery@host1: cache_reloaded on celery@host1
  celery@host2: cache_reloaded on celery@host2
Pinged 2 workers:
  celery@host1 - 1719561600.0
  celery@host2 - 1719561600.1

Targeted Broadcast

from celery import Celery

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

def reload_worker_config(worker_names=None, config_key=None, config_value=None):
    arguments = {}
    if config_key and config_value:
        arguments['config_key'] = config_key
        arguments['config_value'] = config_value

    result = app.control.broadcast(
        'reload_config',
        arguments=arguments,
        reply=True,
        destination=worker_names,
    )
    return result

result = reload_worker_config(
    worker_names=['celery@worker1', 'celery@worker3'],
    config_key='task_soft_time_limit',
    config_value='600',
)
print(f"Config updated on targeted workers: {list(result.keys())}")

Expected output:

Config updated on targeted workers: ['celery@worker1', 'celery@worker3']

Broadcast with Arguments

from celery import Celery
from celery.worker.control import Panel
import json

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

@Panel.register
def clear_rate_limits(panel, pattern=None, **kwargs):
    cleared = []
    for task_name, limit in list(app.conf.task_routes.items()):
        if pattern is None or pattern in task_name:
            cleared.append(task_name)
            app.control.rate_limit(task_name, None)
    return {
        'worker': panel.hostname,
        'cleared_tasks': cleared,
        'count': len(cleared),
    }

result = app.control.broadcast(
    'clear_rate_limits',
    arguments={'pattern': 'email'},
    reply=True,
)
print(f"Cleared rate limits on {len(result)} workers")
for worker, reply in result.items():
    print(f"  {worker}: {reply['count']} tasks cleared")

Expected output:

Cleared rate limits on 2 workers
  celery@host1: 3 tasks cleared
  celery@host2: 3 tasks cleared

Common Mistakes

• Forgetting reply=True when expecting responses -- without reply=True, broadcast sends the command but returns immediately with empty results. Add reply=True and a reasonable timeout for response collection.
• Timeout too short for large clusters -- broadcast must collect replies from all workers within the timeout. For 100+ workers, set timeout to at least 30 seconds to account for network latency.
• Not handling worker unavailability -- if a worker is stopped or unreachable, its reply is missing. Check which workers replied and which did not. Do not assume all workers received the command.
• Overusing broadcast for per-worker tasks -- broadcast sends to every worker. If only one worker needs the command, use direct control or a targeted task. Broadcast creates unnecessary broker traffic at scale.
• Missing @Panel.register decorator -- custom handlers must be decorated and accessible at module level. The module must be imported when the worker starts. Use imports in the worker startup phase.

Practice Questions

1. How does broadcast differ from a regular task?
2. What is the purpose of the @Panel.register decorator?
3. How do you target specific workers with a broadcast?
4. What happens if a worker does not respond to a broadcast within the timeout?
5. How would you implement a rolling restart using broadcast?

Challenge

Build a cluster management system using broadcast: (1) register handlers for drain (stop accepting new tasks), undrain (resume accepting), and reload_config, (2) implement a rolling update that drains workers one at a time, updates their config, and undrains them, (3) verify all workers received the update by checking reply count equals expected count, (4) implement a broadcast health check that reports worker metrics, and (5) handle worker timeouts with retry logic.

FAQ

Mini Project

Build a cluster coordination hub that: (1) registers handlers for cache invalidate, config reload, task drain, and log level change, (2) provides a web UI with buttons for each operation and optional worker targeting, (3) collects and displays replies from all workers with timestamps, (4) implements an audit log of all broadcast commands sent with requester identity, and (5) includes a dry-run mode that shows which workers would be affected without executing.

What's Next

Continue with Worker Inspection to learn how to inspect worker state remotely. Then explore Custom Task Classes for advanced task behavior customization.