Priority Queue — Complete Guide

In this tutorial, you will learn about Priority Queue. We cover key concepts, practical examples, and best practices to help you master this topic.

A priority queue ensures high-priority messages are processed before lower-priority ones, enabling time-sensitive operations to skip the line ahead of routine tasks.

What You'll Learn

By the end of this lesson, you will understand how priority queues work, how to configure priority in RabbitMQ, how to implement priority with multiple queues, and when to use priority-based message processing.

Why It Matters

Not all messages are equally urgent. A payment cancellation must be processed immediately, but a profile picture update can wait. Without priorities, urgent messages wait behind routine ones, causing delays that affect user experience and business operations.

Real-World Use

A customer support platform processes tickets. Urgent "account suspended" tickets must be handled within minutes. Routine "change password" requests can wait hours. Priority queues ensure urgent tickets jump to the front of the queue.

Priority Queue Architecture

flowchart LR
    P[Producer] --> Q[Priority Queue]
    Q --> H{High Priority?}
    H -->|Yes| C1[Consumer picks first]
    H -->|No| C2[Processed after high]
    style Q fill:#f90,color:#fff

Messages with higher priority are delivered before messages with lower priority. Within the same priority level, FIFO order is preserved.

RabbitMQ Priority Queue

import pika
import json

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

args = {'x-max-priority': 10}
channel.queue_declare(queue='task_queue', durable=True, arguments=args)

tasks = [
    ('Low priority task', 1),
    ('High priority task', 10),
    ('Medium priority task', 5),
    ('Critical task', 10),
]

for task_text, priority in tasks:
    channel.basic_publish(
        exchange='',
        routing_key='task_queue',
        body=task_text,
        properties=pika.BasicProperties(
            delivery_mode=2,
            priority=priority
        )
    )
    print(f"Published: {task_text} (priority {priority})")

connection.close()

Expected output:

Published: Low priority task (priority 1)
Published: High priority task (priority 10)
Published: Medium priority task (priority 5)
Published: Critical task (priority 10)

# Consumer
import pika

conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.queue_declare(queue='task_queue', durable=True)

def callback(ch, method, properties, body):
    prio = properties.priority if properties.priority else 0
    print(f"Processing (priority {prio}): {body.decode()}")
    ch.basic_ack(delivery_tag=method.delivery_tag)

ch.basic_consume(queue='task_queue', on_message_callback=callback)
print("Waiting for tasks...")
ch.start_consuming()

Expected output:

Waiting for tasks...
Processing (priority 10): High priority task
Processing (priority 10): Critical task
Processing (priority 5): Medium priority task
Processing (priority 1): Low priority task

High-priority messages (10) are delivered before medium (5) and low (1).

Multi-Queue Priority Pattern

For brokers without native priority support, use separate queues per priority level:

import pika
import json

conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()

queues = {'high': 10, 'medium': 5, 'low': 1}
for name, prio in queues.items():
    ch.queue_declare(queue=f'priority_{name}', durable=True)

ch.basic_publish(exchange='', routing_key='priority_high',
                 body='URGENT! Server is down', properties=pika.BasicProperties(delivery_mode=2))
ch.basic_publish(exchange='', routing_key='priority_low',
                 body='Update profile picture', properties=pika.BasicProperties(delivery_mode=2))
ch.basic_publish(exchange='', routing_key='priority_medium',
                 body='Generate weekly report', properties=pika.BasicProperties(delivery_mode=2))

print("Published to priority queues")
conn.close()

Consumer with priority ordering:

import pika, time

conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()

priority_queues = ['priority_high', 'priority_medium', 'priority_low']
for q in priority_queues:
    ch.queue_declare(queue=q, durable=True)

def callback(ch, method, properties, body):
    print(f"[{method.routing_key}] {body.decode()}")
    ch.basic_ack(delivery_tag=method.delivery_tag)

for q in priority_queues:
    ch.basic_consume(queue=q, on_message_callback=callback)

ch.start_consuming()

Priority Use Cases

Common Mistakes

1. Using Too Many Priority Levels

More than 5-10 priority levels add complexity without benefit. Consumers must handle each level. Stick to 3-5 levels: critical, high, medium, low, and background.

2. Starving Low-Priority Messages

If high-priority messages keep arriving, low-priority ones never get processed. Implement starvation prevention: increase the priority of waiting messages or allocate minimum processing time to each level.

3. Relying on Priority in Multiple-Consumer Scenarios

Priority queues work best with a single consumer. With multiple consumers, high-priority messages are delivered first, but consumers may Process them in unpredictable order. Use separate queues for strict priority.

4. Not Setting Max Priority on the Queue

In RabbitMQ, you must set x-max-priority when declaring the queue. Without it, the priority field is ignored. The max priority default is 0 (no priority).

5. Combining Priority with Ordering

Priority queues break FIFO ordering by design. If you need both ordering and priority, use multiple queues per priority level and process them in order.

Practice Questions

1. How does RabbitMQ implement priority queues?

Set x-max-priority on queue declaration. Messages with higher priority values are delivered before lower-priority ones. Within the same priority, FIFO order is preserved.

2. What is the risk of priority queues?

Starvation. If high-priority messages arrive continuously, low-priority messages may never be processed. Implement fairness mechanisms to prevent this.

3. How do you implement priority without broker support?

Use separate queues per priority level. The consumer checks high-priority queues first, then medium, then low. This is called priority queuing with multiple queues.

4. What priority value range should I use?

1-10 is sufficient for most applications. Higher ranges add granularity but increase complexity. Use odd numbers to leave room for future priority levels.

Challenge

Design a priority system for an incident management platform: P1 (critical, <1min), P2 (high, <5min), P3 (medium, <30min), P4 (low, <4hrs). Implement starvation prevention so P4 incidents eventually get processed even during P1 storms.

FAQ

Mini Project: Priority-Based Task Scheduler

import pika
import json
import threading
import time
import random

def priority_consumer():
    conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    ch = conn.channel()
    ch.queue_declare(queue='priority_tasks', durable=True, arguments={'x-max-priority': 10})

    def cb(c, m, p, body):
        prio = p.priority or 0
        duration = random.uniform(0.1, 0.5)
        print(f"[P{prio}] Processing: {body.decode()} (takes {duration:.1f}s)")
        time.sleep(duration)
        c.basic_ack(delivery_tag=m.delivery_tag)

    ch.basic_qos(prefetch_count=1)
    ch.basic_consume(queue='priority_tasks', on_message_callback=cb)
    ch.start_consuming()

t = threading.Thread(target=priority_consumer, daemon=True)
t.start()

time.sleep(1)
conn = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
ch = conn.channel()
ch.queue_declare(queue='priority_tasks', durable=True, arguments={'x-max-priority': 10})

for i in range(10):
    prio = random.randint(1, 10)
    ch.basic_publish(
        exchange='', routing_key='priority_tasks',
        body=f'Task-{i}',
        properties=pika.BasicProperties(priority=prio, delivery_mode=2)
    )

print("Published 10 tasks with random priorities")
conn.close()
time.sleep(5)

Expected output:

Published 10 tasks with random priorities
[P9] Processing: Task-3 (takes 0.3s)
[P10] Processing: Task-7 (takes 0.2s)
[P8] Processing: Task-1 (takes 0.4s)
...

What's Next

Now that you understand priority queues, explore the request-reply pattern for synchronous request-response over Message Queues, then learn about competing consumers for parallel task processing.