AMQP Celery Client

Celery is a Python task queue that uses LavinMQ as its broker. It talks AMQP through Kombu and py-amqp, so a worker only needs a broker URL. Install it with:

pip install celery

Send and run a task

The example reads the broker URL from BROKER_URL and falls back to a local LavinMQ at amqp://guest:guest@localhost//.

Save the configuration as celeryconfig.py:

import os

# Recommended settings for CloudAMQP, see https://www.cloudamqp.com/docs/celery.html
broker_url = os.environ.get("BROKER_URL", "amqp://guest:guest@localhost//")
broker_pool_limit = 1  # Decreases connection usage
broker_heartbeat = None  # CloudAMQP uses TCP keep-alive instead
broker_connection_timeout = 30  # Long timeout, e.g. for slow DNS lookups
result_backend = None  # Don't store results in the broker
event_queue_expires = 60  # Delete celeryev queues without consumers after a minute
worker_prefetch_multiplier = 1  # Fetch one message at a time
worker_concurrency = 50  # For I/O-bound tasks; for CPU-bound tasks, use the number of cores

Save the task as tasks.py:

from celery import Celery
from celery.utils.log import get_task_logger

app = Celery("tasks")
app.config_from_object("celeryconfig")
logger = get_task_logger(__name__)


@app.task
def hello(name):
    message = f"Hello {name}!"
    logger.info(message)
    return message

Save the publisher as publisher.py in the same directory:

from tasks import hello

if __name__ == "__main__":
    hello.delay("world")
    print(" [x] Sent hello('world')")

Start the worker in one terminal and run python publisher.py in another:

celery -A tasks worker --without-heartbeat --without-gossip --without-mingle

The --without-* flags stop the worker from sending heartbeat, gossip and mingle messages, which otherwise add up to hundreds of messages per second.

The full, runnable example lives in python/celery in cloudamqp/examples, and CI tests it against LavinMQ.

Good to know

• Tasks get acked before they run. task_acks_late defaults to False, so a worker that crashes mid-task loses that task. Set task_acks_late = True to ack after the task finishes, and keep tasks idempotent, since LavinMQ then redelivers unfinished ones.
• Prefetch is the multiplier times the concurrency. worker_prefetch_multiplier defaults to 4. The example sets it to 1, so each of the 50 worker processes holds one task at a time.
• Workers reconnect on their own. broker_connection_retry is on by default and gives up after 100 attempts, set by broker_connection_max_retries.
• An amqps:// URL alone skips the hostname check. The certificate gets verified against the system trust store, but not the hostname. Add broker_use_ssl = {"server_hostname": None} to check it against the host in the URL.

The Celery documentation lists every setting, and the Celery task queue tutorial builds a larger app step by step.