Queue Types in Python

The Story That Explains All Queues

The Big Picture — All Queue Types At A Glance

Type 1 — FIFO Queue (First In, First Out)

The default and most intuitive queue. Items enter at the tail and leave from the head, in the exact order they arrived. This is the "coffee shop line" of data structures.

from collections import deque

# deque = the fast Python FIFO (O(1) at both ends)
q = deque()

q.append("A")       # enqueue at tail
q.append("B")
q.append("C")

print(q.popleft())   # "A" — dequeue from head
print(q.popleft())   # "B"
print(q.popleft())   # "C"

Type 2 — LIFO Queue (Stack)

A stack. Items enter and leave from the same end — the top. The last thing you added is the first thing you get back. Perfect for undo/redo history, function-call frames, and depth-first search.

# Option A: Python list as a stack (works perfectly, O(1) at end)
stack = []
stack.append("A")         # push
stack.append("B")
stack.append("C")
print(stack.pop())     # "C" — last in, first out
print(stack.pop())     # "B"

# Option B: collections.deque (also O(1), thread-safe append/pop)
from collections import deque
stack = deque()
stack.append("A")
stack.append("B")
print(stack.pop())     # "B"

# Option C: queue.LifoQueue (thread-safe, blocking API)
from queue import LifoQueue
s = LifoQueue()
s.put("A"); s.put("B"); s.put("C")
print(s.get())        # "C"

Type 3 — Priority Queue

Items don't come out in insertion order. Each item carries a priority, and the item with the highest priority (in Python: the lowest priority number) always comes out first, regardless of when it was added.

# Option A: heapq — the low-level, fast min-heap
import heapq

pq = []
heapq.heappush(pq, (5, "Change ink"))
heapq.heappush(pq, (1, "Heart attack"))
heapq.heappush(pq, (3, "Cough"))
heapq.heappush(pq, (1, "Stroke"))
heapq.heappush(pq, (2, "Broken arm"))

while pq:
    print(heapq.heappop(pq))
# (1, 'Heart attack')
# (1, 'Stroke')
# (2, 'Broken arm')
# (3, 'Cough')
# (5, 'Change ink')

# Option B: queue.PriorityQueue — thread-safe, blocking API
from queue import PriorityQueue
pq = PriorityQueue()
pq.put((5, "low urgency"))
pq.put((1, "urgent!"))
print(pq.get())        # (1, 'urgent!')

Type 4 — Deque (Double-Ended Queue)

Pronounced "deck". You can add or remove from either end in O(1). A deque can act as a FIFO, a LIFO, or both at once. Python's collections.deque is the most versatile queue in the standard library.

from collections import deque

d = deque(["B", "C", "D"])

# Add to either end
d.appendleft("A")      # [A, B, C, D]
d.append("E")           # [A, B, C, D, E]

# Remove from either end
print(d.popleft())     # "A"  ← [B, C, D, E]
print(d.pop())         # "E"  ← [B, C, D]

# Bounded deque — auto-drops from the OPPOSITE end when full
recent = deque(maxlen=3)
for i in range(6):
    recent.append(i)
print(recent)              # deque([3, 4, 5], maxlen=3)

# Rotate — very fast circular shift
d = deque([1, 2, 3, 4, 5])
d.rotate(2)               # deque([4, 5, 1, 2, 3])
d.rotate(-1)              # deque([5, 1, 2, 3, 4])

Type 5 — Circular Queue (Ring Buffer)

A fixed-size queue where the head and tail pointers wrap around the end of the storage array back to the start. It never grows, never shrinks — perfect for fixed-memory environments (embedded systems, audio buffers, network packet rings).

Implementing a Circular Queue From Scratch

class CircularQueue:
    def __init__(self, capacity):
        self.buf   = [None] * capacity
        self.cap   = capacity
        self.head  = 0       # index to read from
        self.tail  = 0       # index to write to
        self.size  = 0

    def enqueue(self, item):
        if self.size == self.cap:
            raise OverflowError("Queue full")
        self.buf[self.tail] = item
        self.tail = (self.tail + 1) % self.cap    # wrap around!
        self.size += 1

    def dequeue(self):
        if self.size == 0:
            raise IndexError("Queue empty")
        item = self.buf[self.head]
        self.buf[self.head] = None
        self.head = (self.head + 1) % self.cap    # wrap around!
        self.size -= 1
        return item

    def __len__(self): return self.size


# Demo
cq = CircularQueue(4)
for ch in "ABCD":
    cq.enqueue(ch)
print(cq.dequeue(), cq.dequeue())    # A B
cq.enqueue("E"); cq.enqueue("F")         # tail wraps back to slot 0
print(cq.buf)                            # ['E', 'F', 'C', 'D']

Shortcut — Use deque(maxlen=…) As A Circular Buffer

from collections import deque

# A ring of the last 5 sensor readings
ring = deque(maxlen=5)
for reading in [10, 11, 13, 12, 14, 15, 16]:
    ring.append(reading)     # old readings drop off the LEFT automatically

print(ring)                # deque([13, 12, 14, 15, 16], maxlen=5)
print(sum(ring) / len(ring))  # rolling average

Python's Queue Module — Thread-Safe Implementations

The four types above are conceptual. When you need thread-safe versions with blocking put/get and size limits, use the queue module.

from queue import Queue, LifoQueue, PriorityQueue, SimpleQueue

# 1. Standard FIFO Queue
q = Queue(maxsize=10)
q.put("A"); q.put("B"); q.put("C")
print(q.get())          # A

# 2. Stack (LIFO)
s = LifoQueue()
s.put("A"); s.put("B"); s.put("C")
print(s.get())          # C — last in, first out

# 3. Priority (lowest number wins)
p = PriorityQueue()
p.put((3, "low"))
p.put((1, "urgent"))
p.put((2, "med"))
print(p.get())          # (1, 'urgent')

# 4. SimpleQueue — no maxsize, no join(), fastest
sq = SimpleQueue()
sq.put("fast")
print(sq.get())         # fast

Multiprocessing Queues — Cross-Process Communication

The queue module only works between threads (same memory space). For processes, use multiprocessing.Queue or JoinableQueue — they pickle and pipe objects across process boundaries.

from multiprocessing import Process, Queue, JoinableQueue

def worker(q):
    while True:
        item = q.get()
        if item is None: break
        print(f"got {item}")
        q.task_done()             # only on JoinableQueue

if __name__ == "__main__":
    q = JoinableQueue()          # like Queue + task_done/join

    p = Process(target=worker, args=(q,))
    p.start()

    for i in range(5):
        q.put(f"task {i}")

    q.join()                     # wait for all task_done()
    q.put(None)                  # poison pill
    p.join()

Asyncio Queue — For Async/Await Code

When you're inside asyncio code, you want asyncio.Queue. It has the same API as queue.Queue, but its put()/get() return coroutines you must await. It's NOT thread-safe — it's coroutine-safe within a single event loop.

import asyncio

async def producer(q, name):
    for i in range(5):
        await asyncio.sleep(0.1)
        await q.put(f"{name}-{i}")

async def consumer(q, name):
    while True:
        item = await q.get()
        if item is None: break
        print(f"[{name}] got {item}")
        q.task_done()

async def main():
    q = asyncio.Queue(maxsize=10)

    prods = [asyncio.create_task(producer(q, f"P{i}")) for i in range(2)]
    cons  = [asyncio.create_task(consumer(q, f"C{i}")) for i in range(3)]

    await asyncio.gather(*prods)
    await q.join()

    for _ in cons:
        await q.put(None)
    await asyncio.gather(*cons)

asyncio.run(main())

heapq — The Raw Priority Queue Building Block

heapq gives you a min-heap on a regular Python list. It's what PriorityQueue uses internally, but without any locking overhead. Fastest choice for single-threaded priority work.

import heapq

# A heap is just a plain list, treated with heap operations
h = []
heapq.heappush(h, 3)
heapq.heappush(h, 1)
heapq.heappush(h, 4)
heapq.heappush(h, 1)
heapq.heappush(h, 5)

print(heapq.heappop(h))       # 1 (smallest first)
print(heapq.heappop(h))       # 1
print(heapq.heappop(h))       # 3

# Turn an existing list into a heap in O(n)
data = [7, 3, 9, 1, 4, 2]
heapq.heapify(data)             # in-place: [1, 3, 2, 7, 4, 9]
print(data)

# Top-K largest / smallest in O(n log k)
print(heapq.nlargest(3, data))   # [9, 7, 4]
print(heapq.nsmallest(3, data))  # [1, 2, 3]

# Priority queue with tie-breaker (avoids TypeError on equal priorities)
import itertools
counter = itertools.count()
pq = []
heapq.heappush(pq, (1, next(counter), "first urgent"))
heapq.heappush(pq, (1, next(counter), "second urgent"))
heapq.heappush(pq, (3, next(counter), "normal"))
while pq:
    print(heapq.heappop(pq)[2])

Bounded vs Unbounded Queues — The Backpressure Diagram

Real-World Scenarios — Which Queue For What

Practical Example — Multi-Queue Task Scheduler

A worker that pulls from THREE queues in strict priority order: emergency tasks first, then normal tasks, then background maintenance.

from queue import Queue
import threading, time, random

emergency  = Queue()
normal     = Queue()
background = Queue()

def worker():
    while True:
        # Try each queue in priority order
        for name, q in [("emergency", emergency),
                        ("normal",    normal),
                        ("bg",        background)]:
            try:
                task = q.get_nowait()
                if task is None: return
                print(f"[{name}] processing {task}")
                time.sleep(0.1)
                break
            except:
                continue
        else:
            time.sleep(0.05)     # all empty, wait a bit

# Kick off a few workers
threads = [threading.Thread(target=worker) for _ in range(3)]
for t in threads: t.start()

# Feed mixed tasks
for i in range(10): background.put(f"bg-{i}")
for i in range(5):  normal.put(f"task-{i}")
for i in range(2):  emergency.put(f"fire-{i}")

time.sleep(3)
for _ in threads: emergency.put(None)
for t in threads: t.join()

Complete Comparison Table — All Queue Types

*A Python list is atomic for single append/pop from the same end, but mixed operations from multiple threads can still race.

Decision Flowchart — Which Queue Should I Use?

Golden Rules