How to build a job queue in Python with RabbitMQ

If you have ever had a Python script that works fine until it has too much to do, this is the tutorial for you.

We are going to build a job queue: one script that hands out work, and several workers that do it at the same time. By the end you will have something you can run on your laptop, and you will be able to add more workers to make it faster without changing a line of logic.

Here is the shape of it:

send_jobs.py ──▶ [ RabbitMQ queue ] ──┬──▶ worker.py
                                      ├──▶ worker.py
                                      └──▶ worker.py

What you need: Python 3.8+, and Docker (only to run RabbitMQ no configuration required).

Step 1: Start RabbitMQ

One command:

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3-management

Give it about twenty seconds, then open http://localhost:15672 and log in with guest / guest. That is RabbitMQ's admin dashboard. Keep the tab open you will watch your jobs move through it later, which is by far the easiest way to tell what is happening.

Port 5672 is what your code talks to. Port 15672 is just the dashboard.

Step 2: Install the Python client

pip install pika

pika is the standard RabbitMQ library for Python.

Step 3: Send some jobs

Create send_jobs.py:

import json
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()

# create the queue (safe to call every time it won't duplicate)
channel.queue_declare(queue="tasks", durable=True)

for task_id in range(1, 11):
    message = json.dumps({"task_id": task_id, "seconds": 2})

    channel.basic_publish(
        exchange="",
        routing_key="tasks",                 # which queue to put it in
        body=message,
        properties=pika.BasicProperties(delivery_mode=2),
    )
    print(f"Queued task {task_id}")

connection.close()

Run it:

python send_jobs.py

Now go back to the dashboard, open the Queues tab, and you will see tasks holding 10 messages. Nothing has processed them yet, and that is the point the jobs are sitting safely in the queue waiting for someone to pick them up.

Two things worth knowing about what you just wrote:

• durable=True means the queue survives a RabbitMQ restart.
• delivery_mode=2 means the messages themselves get written to disk instead of only living in memory.

You need both. One without the other still loses data.

Step 4: Process the jobs

Create worker.py:

import json
import time
import pika

connection = pika.BlockingConnection(pika.ConnectionParameters("localhost"))
channel = connection.channel()

channel.queue_declare(queue="tasks", durable=True)

# only give this worker one job at a time
channel.basic_qos(prefetch_count=1)

def handle_task(ch, method, properties, body):
    task = json.loads(body)
    print(f"  started task {task['task_id']}")

    time.sleep(task["seconds"])          # pretend this is real work

    print(f"  finished task {task['task_id']}")
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(queue="tasks", on_message_callback=handle_task)

print("Waiting for tasks. Press CTRL+C to stop.")
channel.start_consuming()

Run it:

python worker.py

It will chew through all 10 jobs, two seconds each, then sit and wait for more. Leave it running.

Step 5: Now the good part: add another worker

Open a second terminal and start another worker:

python worker.py

Now open a third terminal and send the jobs again:

python send_jobs.py

Watch both worker terminals. RabbitMQ hands out the jobs one per worker, alternating, and the whole batch finishes in roughly half the time. Start a third worker and it gets faster again.

You did not change any logic to make that happen. You started another process.

That is the entire trick, and it scales a lot further than a laptop demo suggests it is the same pattern behind a pipeline I took from 40,000 to over 2 million records a day on the same hardware.

About that prefetch_count=1 line: it tells RabbitMQ to give a worker its next job only when it has finished the current one. Without it, RabbitMQ pushes a big batch at whichever worker connects first, and that worker sits on a pile of jobs while the others do nothing. Keep it at 1 while you are learning.

Step 6: Make sure a crash doesn't eat a job

Right now, try this. Start one worker, send the jobs, and while it is mid-task press Ctrl+C twice to kill it hard. Then start it again.

The job it was working on comes back and gets processed. Nothing was lost.

That works because of one line:

ch.basic_ack(delivery_tag=method.delivery_tag)

An ack is your worker telling RabbitMQ "this one is genuinely done,you can forget it." Until that happens, RabbitMQ holds onto the message. If your worker dies, the unacknowledged job goes back on the queue for someone else.

Notice where that line is: at the end of handle_task, after the work. This is the single most important detail in the whole tutorial.

Here is the version that quietly loses data:

# DON'T DO THIS
channel.basic_consume(
    queue="tasks",
    on_message_callback=handle_task,
    auto_ack=True,          # acknowledges before your code has run
)

With auto_ack=True, RabbitMQ marks the job done the moment it hands it over before handle_task has done anything. If the worker crashes during the job, the job is gone, and nothing anywhere logs an error.

Good news: pika defaults to auto_ack=False, so you are safe unless you turn it on. Plenty of tutorials turn it on to "keep the example simple." Don't copy those.

Step 7: Handle jobs that can never succeed

Some jobs will fail every single time bad input, a bug, a service that is gone for good. If you just put those back on the queue, they loop forever and block everything behind them.

The fix is a dead letter queue: a second queue where hopeless jobs go to be looked at later.

# a place for failed jobs to land
channel.exchange_declare("dlx", exchange_type="direct", durable=True)
channel.queue_declare("tasks.dead", durable=True)
channel.queue_bind("tasks.dead", "dlx", routing_key="tasks")

# tell the main queue where to send rejects
channel.queue_declare(
    queue="tasks",
    durable=True,
    arguments={
        "x-dead-letter-exchange": "dlx",
        "x-dead-letter-routing-key": "tasks",
    },
)

Then, when a job is genuinely unrecoverable, reject it instead of acking it:

ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

requeue=False is what sends it to the dead letter queue rather than back around the loop.

⚠️ A gotcha that will catch you: you cannot add those arguments to a queue that already exists. If you already ran the earlier code, RabbitMQ will throw PRECONDITION_FAILED. Delete the tasks queue in the dashboard first (Queues → tasks → Delete), then run the new code.

What you have now

A queue you can throw work at, workers you can add for more throughput, jobs that survive a crash, and somewhere for failures to go. That is a real background-job system, in about sixty lines.

Some things to try from here:

• Replace time.sleep() with something real an API call, an image resize, an email send.
• Run the workers on two different machines pointed at the same RabbitMQ.
• Watch the Queues tab while you send a big batch. Seeing the depth rise and drain teaches you more about how this behaves than any article will.

If you want the production version of this how to pick a prefetch value, autoscaling workers on queue depth instead of CPU, and when you should reach for Kafka instead of RabbitMQ I wrote that up here: RabbitMQ Data Pipelines: Scale ETL without adding compute.