Python Worker Message Consumption

Your API just went viral. Thousands of requests per second are hitting your server, and every one of them needs time-consuming work done: resizing images, sending emails, updating search indexes. If you handle all of that inline, your users wait, your database chokes, and your server dies. The solution: offload the work to a queue and process it asynchronously. But popping a message off a queue is only half the story — you need a reliable, scalable worker that consumes those messages and does the job without losing data or crashing under load. This lesson shows you exactly how to build one in Python.

The problem this lesson solves

Synchronous processing is simple, but it doesn't scale. When your web application performs heavy tasks in the request/response cycle, you're tying up precious server resources and making your users wait for work that isn't critical to their response. Think about uploading a profile photo: the user doesn't need to wait for the thumbnail to be generated — they just need to see a success message. But if that thumbnail generation blocks the request, you've added hundreds of milliseconds (or worse) to every upload.

Queues solve this by decoupling production from consumption. Your API becomes a producer that publishes messages to a queue, and separate consumers pick them up later. This decoupling is powerful, but it shifts the problem: how do you reliably consume messages with a Python worker? A naive loop that calls get() and processes the message is a start, but it's riddled with pitfalls. What happens if your worker crashes mid-processing? What if the queue is empty? What if the same message gets delivered twice? What if your consumer is slower than your producer — do you lose messages or just lag behind?

This lesson gives you the mental model and the practical skills to build a production-grade Python worker that handles these edge cases gracefully. You'll learn not just how to consume, but how to do it right.

Core concept / mental model

Think of your queue as a conveyor belt at a busy kitchen. The chefs (producers) place orders (messages) on the belt. The cooks (workers) stand at the belt, pick up orders, and prepare them. The belt itself doesn't do any cooking — it just holds items temporarily. Your Python worker is the cook: its job is to pick up a message, process it, and confirm it's done.

The critical detail is the acknowledgment. In most message systems, the queue doesn't automatically remove a message when a consumer reads it. Instead, the consumer must explicitly tell the queue "I've finished this, you can delete it." This is called an ack (acknowledgment). If the worker crashes before sending the ack, the message goes back to the queue and will be redelivered to another worker. This is the foundation of at-least-once delivery: your message will never be lost, but it might be delivered more than once.

This leads to a key realization: you must design your worker to be idempotent, meaning processing the same message twice produces the same result. If you're charging a customer's credit card, you don't want to charge them twice. If you're sending a welcome email, you don't want to send it twice. Idempotency is the price you pay for reliability.

Here's a mental diagram of the flow:

Producer → Message Queue → Consumer (Python Worker)
                                 │
                                 ├── process()
                                 └── ack()

The queue is a buffer between producer and consumer. It absorbs spikes in traffic and allows the consumer to work at its own pace.

How it works step by step

Let's walk through the lifecycle of a message from production to successful consumption:

1. Producer publishes the message. The producer connects to the queue (e.g., RabbitMQ, Redis, or Amazon SQS) and sends a serialized message, usually JSON.

2. Message sits in the queue. The broker stores it until a consumer is ready. It may persist to disk, depending on your broker's durability settings.

3. Consumer polls for a message. Your Python worker calls a method like get() or receive() to pull a message. This can be a blocking call (waits until a message arrives) or a non-blocking call (returns immediately even if empty).

4. Consumer processes the message. Your worker executes the business logic — resizing an image, sending an email, updating a database.

5. Consumer sends an acknowledgment. After successful processing, the worker tells the broker to delete the message. This is the ack() call.

6. If processing fails: the worker can send a nack (negative acknowledgment) or simply not ack, causing the message to be redelivered. Some brokers support a dead-letter queue after a maximum number of retries.

Step 4 is where all your business logic lives. It's where you'll call external APIs, write to databases, or trigger other services. The rest is plumbing.

The word worker is important here — it's not a full-fledged service, but a lightweight, dedicated process that just consumes and processes messages. You might run multiple copies of the same worker to scale throughput.

Hands-on walkthrough

Let's build a simple but robust Python worker. We'll use Redis with the redis-py library as our queue, because it's simple, widely used, and gives you fine control over the mechanics. (We'll compare other brokers in the next section.)

First, install the library:

pip install redis

Make sure you have Redis running locally (e.g., docker run -p 6379:6379 redis).

Now, let's write a worker loop that simulates processing tasks like image resizing or sending emails.

import json
import time
import redis

# Connect to Redis
r = redis.Redis(host='localhost', port=6379, db=0)

QUEUE_KEY = 'my_queue'

def process_message(data: dict) -> None:
    """Simulate a real task—e.g., resize an image or send an email."""
    task_type = data.get('type')
    task_id = data.get('id')
    print(f"Processing {task_type} task {task_id}")
    # Simulate work
    time.sleep(0.5)
    # If a task fails, raise an exception to trigger a nack
    if data.get('fail'):
        raise RuntimeError("Simulated failure")
    print(f"Completed {task_type} task {task_id}")

def main():
    print("Worker started. Listening for messages...")
    while True:
        # Block for up to 1 second, listening for a raw tuple (queue_name, message_json)
        item = r.blpop(QUEUE_KEY, timeout=1)
        if item is None:
            # No message, keep waiting
            continue
        _, message_json = item
        try:
            data = json.loads(message_json)
            process_message(data)
            # Success: nothing to do, blpop already removed it
        except Exception as e:
            print(f"Error processing message {message_json}: {e}")
            # Re-push the message to the back of the queue for a retry
            # Better: use a retry counter and a dead-letter queue
            r.rpush(QUEUE_KEY, message_json)

if __name__ == "__main__":
    main()

Expected output when you run this worker and then publish a few messages:

Worker started. Listening for messages...
Processing email task 123
Completed email task 123
Processing image task 456
Error processing message {"type": "image", "id": 456, "fail": true}: Simulated failure

This is a minimal worker. It uses blpop (blocking left pop) to atomically get and remove the message. If processing fails, we re-push the message to the back of the queue — a simple retry mechanism. But notice a flaw: if the worker crashes before processing (e.g., after blpop but before the try), the message is already gone and lost forever. This is where the ack pattern becomes crucial.

To get true at-least-once delivery, you need a queue system that separates get from ack. Let's mimic that with a two-step approach using Redis lists:

import json
import time
import redis

r = redis.Redis(host='localhost', port=6379, db=0)

QUEUE_KEY = 'my_queue'
IN_PROGRESS_KEY = 'my_queue_in_progress'

def get_message():
    """Atomically move a message from the main queue to the in-progress list."""
    # RPOPLPUSH: pop from the right, push to the left of another list
    msg = r.rpoplpush(QUEUE_KEY, IN_PROGRESS_KEY)
    return msg

def process_message(message: str):
    data = json.loads(message)
    print(f"Processing task {data['id']}")
    time.sleep(0.5)
    # Simulate a random failure
    if data.get('fail'):
        raise RuntimeError("Process failed")

def ack(message: str):
    """Remove the message from the in-progress list on success."""
    r.lrem(IN_PROGRESS_KEY, 1, message)

def nack(message: str):
    """On failure, push the message back to the main queue."""
    r.lpush(QUEUE_KEY, message)
    r.lrem(IN_PROGRESS_KEY, 1, message)

while True:
    msg = get_message()
    if msg is None:
        time.sleep(0.1)
        continue
    try:
        process_message(msg)
        ack(msg)
        print(f"Acked: {msg}")
    except Exception as e:
        print(f"Nack: {msg} due to {e}")
        nack(msg)

Now the message is safe: it's removed from the main queue only after we pull it, but it's moved to an in-progress list so we don't lose it if the worker crashes. This pattern is the basis of how libraries like Celery work.

Compare options / when to choose what

You have many options for the queue broker and the library that wraps it. Here's a comparison to help you choose:

When to choose what:

• If you're building a small service and want to understand the underlying mechanics, use Redis with the rpoplpush pattern.
• If you need message priority, complex routing, or strict delivery guarantees, RabbitMQ is your friend.
• If you're on AWS and want zero maintenance, SQS is the path.
• If you're building a large Django app, Celery is the industry standard.

Variation: using libraries like rq — a simple Python library that wraps Redis. It gives you a ready-made worker and job class, so you don't implement the ack pattern from scratch. Great for quick wins.

Troubleshooting & edge cases

Common issues you'll hit when consuming messages with a Python worker:

• Message lost on crash — If you use blpop directly, a crash after pop loses the message. Fix: use a two-step get/ack approach or a broker that supports it (RabbitMQ, SQS).
• Duplicate processing — Even with acks, you can get duplicates (e.g., a network timeout after processing but before ack). Design your worker to be idempotent — use a unique message ID and check a database before reprocessing.
• Worker stalls on a poisoned message — A message that always fails (e.g., malformed JSON) will be retried forever, blocking your queue. Solution: implement a retry limit and move failed messages to a dead-letter queue.
• Slow consumer, fast producer — The queue backlog grows. Autoscale your workers or add a concurrency mechanism (e.g., threading or asyncio).
• Redis BLPOP timeout — If you use a timeout of 0, it blocks forever; if you use a positive timeout, your worker wakes up periodically. Make sure your loop handles timeouts gracefully.
• Connection drops — Always use connection retries and reconnect logic. Use libraries that handle reconnection (like redis-py does) or wrap your loop in a try/except with a time.sleep.

Pro tip: Always set a visibility timeout or message TTL on your queue to prevent a crashed worker from holding a message indefinitely (SQS has this built-in).

What you learned & what's next

You've just mastered the core of consuming messages with a Python worker. You now understand the problem of synchronous processing, the mental model of a queue conveyor belt, the step-by-step lifecycle of a message, and the critical importance of acks and idempotency. You've built a basic worker and improved it with an ack pattern, and you know how to choose between Redis, RabbitMQ, SQS, and Celery. You're also aware of common pitfalls like message loss, duplicates, and poisoned messages.

This is the foundation for more advanced messaging patterns. In the next lesson of this track, you'll explore idempotency in depth — how to design consumers that can safely reprocess the same message without side effects. You'll also touch on dead-letter queues and how to choreograph retries. You're building a solid foundation that will let you handle production-scale workloads with confidence.