Celery vs Dramatiq vs Arq: Picking a Python Task Queue for Async APIs

How Celery, Dramatiq and Arq execute tasks

A task queue moves slow or failure-prone work (sending email, calling third-party APIs, generating reports) out of the request path. The API serializes a job onto a broker and returns immediately, and separate worker processes pick the job up later. Celery is the long-established option, with many brokers and pool types. Dramatiq is a smaller library built around RabbitMQ or Redis, with fewer settings to tune. Arq is Redis-only and built on asyncio.

The main difference is what a worker actually executes:

CopyCelery / Dramatiq                      Arq
API ──▶ Broker ──▶ Worker process      API ──▶ Redis ──▶ Worker process
                   ├─ thread: def f()                    └─ event loop
                   ├─ thread: def g()                       ├─ await f()
                   └─ thread: def h()                       └─ await g()

Celery and Dramatiq call plain synchronous functions. Each process or thread runs one task at a time.

Arq awaits async def functions on a single event loop per worker. While one task waits on I/O, the loop runs other tasks.

This matters when your code already uses async database drivers and HTTP clients. In Arq, you open a connection pool once in the worker's startup hook and reuse it across tasks.

In Celery or Dramatiq, you have two ways to call async code, and both have costs:

• Wrap each call with asyncio.run. This creates and destroys an event loop per task. Pools bound to the old loop break, and connections can leak.
• Keep a long-lived loop per worker. This avoids the per-task loop, but it is extra lifecycle code you have to write and maintain.

In both cases, a thread blocks for the whole task anyway, so you lose the concurrency benefit that made the async drivers worth using.

Copy# Two separate modules shown together for comparison.
import asyncio

import httpx
from arq.connections import RedisSettings
from celery import Celery

# --- Arq: one event loop per worker, shared client opened at startup ---
async def startup(ctx):
    ctx["http"] = httpx.AsyncClient(timeout=10)

async def shutdown(ctx):
    await ctx["http"].aclose()

async def fetch_status(ctx, url):
    # While this awaits the response, the loop runs other queued tasks
    resp = await ctx["http"].get(url)
    return resp.status_code

class WorkerSettings:  # run with: arq module.WorkerSettings
    functions = [fetch_status]
    on_startup = startup
    on_shutdown = shutdown
    redis_settings = RedisSettings(host="localhost", port=6379)

# --- Celery: sync task that wraps async code per call ---
app = Celery("tasks", broker="redis://localhost:6379/0")

async def _fetch(url):
    # Client must be created inside this loop; a module-level client
    # bound to an earlier loop breaks or leaks connections
    async with httpx.AsyncClient(timeout=10) as client:
        resp = await client.get(url)
        return resp.status_code

@app.task
def fetch_status_celery(url):
    # New event loop per task; the worker thread blocks until it finishes
    return asyncio.run(_fetch(url))

The same HTTP task in Arq, which reuses one async client across tasks on the worker's event loop, and in Celery, which must create a fresh event loop and client for every task with asyncio.run.

Enqueuing jobs from FastAPI without blocking the event loop

The producer side has the same split. Arq's client is native asyncio: you create a pool once with create_pool and call await pool.enqueue_job(...) from the handler. The Redis write yields to the event loop while it waits on the network, so other requests keep moving.

Celery's delay()/apply_async() and Dramatiq's send() are synchronous. Each call serializes the message and publishes it to the broker with blocking socket I/O. Inside an async def endpoint, that call stalls the entire event loop, not just the current request. A one-millisecond publish rarely shows up in testing. Under load, or when the broker is slow, reconnecting or applying backpressure, every in-flight request on that worker waits behind it. The result is tail latency that has nothing to do with the endpoint's own work.

There are two ways to fix this:

• Offload the call. Wrap the enqueue in asyncio.to_thread or Starlette's run_in_threadpool. The loop stays free, but the thread pool is bounded, so a stalled broker can still exhaust it.
• Use a sync endpoint. Declare the endpoint with plain def. FastAPI then runs it in that same thread pool automatically.

Whichever library you use, create the client once, not per request. In FastAPI's lifespan handler, open the Arq pool and store it on app.state, then close it after the yield. For Celery and Dramatiq, configure the app or broker at import or startup so their internal connection pools are reused across requests. Close them on shutdown so you don't leak broker connections during rolling deploys.

Copyimport asyncio
from contextlib import asynccontextmanager

import dramatiq
from arq import create_pool
from arq.connections import RedisSettings
from celery import Celery
from dramatiq.brokers.redis import RedisBroker
from fastapi import FastAPI, Request

# Sync clients: build once at import so their connection pools are shared across requests.
celery_app = Celery("tasks", broker="redis://redis.internal:6379/0")
dramatiq_broker = RedisBroker(url="redis://redis.internal:6379/1")
dramatiq.set_broker(dramatiq_broker)

@dramatiq.actor
def resize_image(image_id: str) -> None: ...  # body runs on the worker, not here

@asynccontextmanager
async def lifespan(app: FastAPI):
    # One async Arq pool for the life of the process.
    app.state.arq = await create_pool(RedisSettings(host="redis.internal", port=6379))
    yield
    # Release broker connections on shutdown so rolling deploys don't leak them.
    await app.state.arq.close()
    celery_app.close()
    dramatiq_broker.close()

app = FastAPI(lifespan=lifespan)

@app.post("/reports/{report_id}")
async def enqueue(report_id: str, request: Request):
    # Arq: native asyncio, the Redis write yields to the event loop.
    await request.app.state.arq.enqueue_job("build_report", report_id)
    # Celery and Dramatiq publish with blocking socket I/O: run them in a worker thread.
    await asyncio.to_thread(celery_app.send_task, "tasks.build_report", args=[report_id])
    await asyncio.to_thread(resize_image.send, report_id)
    return {"queued": report_id}

A FastAPI app that creates one Arq pool, one Celery app and one Dramatiq broker at startup, awaits Arq's async enqueue directly, offloads the synchronous Celery and Dramatiq publishes with asyncio.to_thread, and closes all three clients in the lifespan handler.

Brokers, result backends and delivery guarantees

Celery delivers through Kombu, so it can use RabbitMQ, Redis or SQS as a broker and store results in Redis, a SQL database and several other backends. Dramatiq supports RabbitMQ and Redis, with results available through optional middleware. Arq uses Redis for everything: the queue, job state and results.

Durability depends on the broker. RabbitMQ with durable queues and persistent messages survives restarts. Redis survives a restart only if it is configured to persist. RDB snapshots lose everything written since the last snapshot. AOF with everysec fsync can still lose about a second of writes. Eviction is a separate risk. With maxmemory-policy set to anything other than noeviction, Redis can delete queued Arq jobs under memory pressure without raising an error.

Celery's early ack means a worker crash loses the task. Enabling task_acks_late fixes that, but a task can then run twice. Prefetching makes this worse. Each worker reserves several messages ahead, and when that worker dies, every message it reserved is redelivered. On Redis, a task that runs longer than visibility_timeout is also handed to another worker while the first is still running it.

Every broker here guarantees at-least-once delivery at best, so every task must be idempotent. For example, a charge task should receive a key such as charge:order-8812. Before doing any work, the task runs a Redis SET key NX EX and exits if the key already exists. It also passes the same key to the payment provider, so the provider rejects a duplicate charge even if the task crashes after charging but before recording the result. Apply the same approach to emails by keying on the message ID, which stops a redelivered task from sending the email twice.

Copyimport redis
from celery import Celery

app = Celery("billing", broker="redis://localhost:6379/0")
r = redis.Redis(host="localhost", port=6379, db=1, decode_responses=True)

def charge_card(order_id: str, amount_cents: int) -> None:
    """Placeholder for the real payment call."""

# Late ack: a worker crash means redelivery, so this task can run more than once.
@app.task(bind=True, acks_late=True, max_retries=5)
def charge_order(self, order_id: str, amount_cents: int) -> str:
    key = f"charge:{order_id}"  # same key for every delivery of this order

    # Atomically claim the key. Only one execution can create it.
    # The TTL releases the claim if this worker dies mid-charge.
    claimed = r.set(key, "in-progress", nx=True, ex=600)

    if not claimed:
        if r.get(key) == "done":
            return "duplicate delivery, already charged"  # safe no-op
        # Another execution holds the claim. Check again later
        # instead of charging a second time.
        raise self.retry(countdown=30)

    try:
        # Also pass the key to the provider if it supports idempotency keys.
        charge_card(order_id, amount_cents)
    except Exception:
        r.delete(key)  # release the claim so a retry can try again
        raise

    # Keep the done marker longer than any possible redelivery window.
    r.set(key, "done", ex=7 * 24 * 3600)
    return "charged"

A late-acked Celery task uses a Redis SET NX EX claim on an idempotency key such as charge:order-8812, so a redelivered message cannot charge the same order twice.

Retries, timeouts and what happens when a task hangs

Each library puts retry policy in a different place. In Celery it lives on the task decorator: autoretry_for lists the exceptions to catch, retry_backoff and retry_jitter spread retries out, and max_retries caps them. For more control you can call self.retry(countdown=...) yourself. Dramatiq handles retries in its Retries middleware, which is on by default. You tune it per actor with max_retries, min_backoff and max_backoff, and backoff is exponential. Arq only retries when the job raises Retry(defer=...) or is cancelled, for example during worker shutdown. Ordinary exceptions are recorded as failures. Worker-level max_tries sets the ceiling, and ctx["job_try"] lets the job compute its own backoff.

Time limits show the difference between the execution models most clearly:

• Celery prefork. soft_time_limit raises SoftTimeLimitExceeded so the task can clean up. time_limit kills the child process and replaces it. A hung task costs you one process slot.
• Dramatiq. The TimeLimit middleware raises an exception asynchronously in the worker thread. Python cannot deliver that exception while a thread is blocked inside a C call or socket read, so set explicit I/O timeouts as well.
• Arq. job_timeout cancels the coroutine, but cancellation only takes effect at an await. A synchronous requests.get or a CPU-heavy loop blocks the event loop, so no timeout fires and every concurrent job in that worker stalls with it. Push blocking work to asyncio.to_thread or a process pool.

Celery's defaults also catch long-running workloads off guard. Tasks are acknowledged before they run, so a crash loses the task. Set acks_late=True so the broker redelivers it. Workers also prefetch several messages per process, which can strand queued jobs behind one slow task. Set worker_prefetch_multiplier=1 to prevent that. With Redis as the broker, keep the visibility timeout longer than your longest task. If a task outlives it, the message is redelivered while the original is still running, and the idempotency keys described above become essential.

Copyimport asyncio, httpx, dramatiq
from celery import Celery
from arq import Retry

# --- Celery: policy lives on the decorator; tune acks/prefetch for long tasks
app = Celery("jobs", broker="redis://localhost:6379/0")
app.conf.task_acks_late = True           # ack after the task finishes, so a crash means redelivery
app.conf.worker_prefetch_multiplier = 1  # don't reserve extra messages behind a slow task

@app.task(autoretry_for=(httpx.TransportError,), retry_backoff=True,
          retry_jitter=True, max_retries=5,
          soft_time_limit=50, time_limit=60)  # soft: raises in task; hard: kills child process
def sync_report(url):
    return httpx.get(url, timeout=10).text

# --- Dramatiq: Retries + TimeLimit middleware (broker set up elsewhere); values in ms
@dramatiq.actor(max_retries=5, min_backoff=1_000, max_backoff=60_000,
                time_limit=60_000)
def render_report(url):
    # explicit I/O timeout: TimeLimit can't interrupt a thread blocked in a socket read
    return httpx.get(url, timeout=10).text

# --- Arq: retries only via Retry; job_timeout cancels at the next await
async def fetch_report(ctx, url):
    try:
        async with httpx.AsyncClient(timeout=10) as client:
            r = await client.get(url)
    except httpx.TransportError:
        raise Retry(defer=ctx["job_try"] * 5)  # job computes its own backoff (seconds)
    # blocking work goes off the event loop so other jobs in this worker keep running
    return await asyncio.to_thread(parse, r.text)

def parse(text):
    return len(text)

class WorkerSettings:
    functions = [fetch_report]
    max_tries = 5     # ceiling across all attempts
    job_timeout = 60  # seconds; only effective if the coroutine reaches an await

The same flaky HTTP job written for Celery, Dramatiq and Arq, showing where each library sets retries, backoff and time limits, plus the Celery late-ack and prefetch settings for long-running tasks.

Workflows and periodic jobs: chains, pipelines and cron

Composing multi-step work

Celery's canvas is the richest of the three. A chain passes each result to the next task, a group fans out in parallel, and a chord runs a callback once every task in a group has finished. Dramatiq offers pipelines and groups with completion callbacks. Arq has no composition primitives, so a job that needs a follow-up enqueues it itself.

Chords carry the most risk because they hold coordination state. The result backend has to track how many header tasks have completed. If one header task fails, the callback does not run unless you handle the error explicitly. If the backend loses or expires that state, the callback never fires and nothing reports it. Once a flow needs branching, human approval, compensation steps or runs lasting days, use a dedicated workflow engine that persists each step. Encoding that in broker messages makes it hard to inspect and resume.

Scheduling pitfalls

Celery beat must run as exactly one instance. If two beat processes run during a deploy, every periodic task fires twice. Dramatiq leaves scheduling to system cron or a scheduling library, so you also own the uniqueness guarantee. Arq derives a job ID for each scheduled run, which stops multiple workers from enqueuing the same run.

Two guards work with any of these queues:

• Take a broker-side lock with a TTL before the task does any work.
• Key idempotency on the schedule slot, for example "report for this date", rather than on the message ID.

Concurrency tuning, observability and engineering cost

Concurrency and failure isolation differ more between these libraries than their APIs do. Celery defaults to a prefork pool, and Dramatiq runs processes multiplied by threads. Both isolate CPU-bound work across processes and let a hung or leaking child be replaced. Arq runs one asyncio loop per worker and caps in-flight jobs with max_jobs. That is cheap for I/O-bound jobs. A single CPU-bound or blocking call, however, stalls every job on that loop, so offload it to an executor or run more worker processes.

To size workers, use Little's law: concurrency = arrival rate × average job duration × headroom.

Illustrative estimate (assumed numbers, not a benchmark):

• Assume 50 jobs/s with a mean of 200 ms of mostly I/O wait: 50 × 0.2 × 2 = 20 slots. That fits one Arq worker with max_jobs=20, or 20 Dramatiq threads.
• Assume 40 ms of pure CPU per job instead: 50 × 0.04 = 2 cores fully busy. Threads and event loops cannot help here, so plan for at least 3–4 processes.

With Dramatiq and Arq, you build dashboards, dead-letter inspection and admin tooling yourself.

Each library carries a different engineering cost:

• Celery has a large configuration surface that is easy to misconfigure. In return, it has the largest ecosystem and the most answered questions. Its canvas primitives and result-backend semantics create the deepest migration lock-in.
• Dramatiq has fewer settings and fewer surprises. It also has a smaller contributor base.
• Arq is small enough to read in an afternoon, which saves time for async FastAPI teams that already use Redis. It also ties you to Redis and to its few maintainers.

Decision checklist: when to use each and when to avoid them

Use the table as a first filter, then adjust for your team.

Team size and on-call tolerance. A team of two or three with a light on-call rotation does better with Arq or Dramatiq. Each has few settings, and when something breaks you can read the source to find out why. Celery is easier to justify when someone owns the worker fleet and can investigate prefetch, visibility timeout and acknowledgement behaviour at 3 a.m.

When to avoid each:

• Arq: avoid it if tasks are CPU-bound or call blocking libraries. A single blocking call stalls every coroutine on that worker.
• Dramatiq: avoid it if you need fan-in workflows beyond simple pipelines and groups. You will end up building your own coordination logic.
• Celery: avoid it for a handful of fire-and-forget jobs. Its configuration surface costs more than it saves.

Alternatives. If workflows run for hours or days, need durable state, or require human approval steps, use a dedicated workflow engine. If all you need is simple fan-out to consumers you already run, a plain broker client publishing to a queue or stream may be enough. It adds no framework at all.

How OneEnv can help

Disclosure: OneEnv is the product behind this blog.

OneEnv (oneenv.io) is a workspace-based governance platform for the things services share — environment variables, database structure, API routes and validation rules. A change to a shared value is staged on a branch, reviewed once, and then every affected service explicitly approves or rejects it before it goes live, so you can see who a change touches before it ships instead of finding out in production.

Applied to celery vs dramatiq vs arq, this is where it helps:

Saves time

• Every service that inherits a shared value is listed in one place, so nobody searches repos and chats to find out who is affected.
• One review per branch replaces coordinating each team separately; each service still keeps its own approve/reject.
• Reviewers see the exact diff and the list of affected services up front, so approval is a decision rather than an investigation.

Saves cost

• Fewer production incidents and rollbacks caused by a shared value, route or column changing under a consumer unannounced.
• Less duplicated configuration to maintain: one shared definition inherited by many services.
• Per-service approve/reject means a bad shared value can be refused where it would break something, before it is live.
• OneEnv has a free plan (1 workspace, unlimited services and members, full governance), so trying it costs nothing.

These are qualitative benefits, not measured results — how much you save depends on how many services share the value.

Want to try it? Start at oneenv.io.