Python provides several different types of queue implementations, each suited for different use cases. This part covers queue.Queue, queue.LifoQueue, and queue.PriorityQueue.
A queue is a data structure that follows the First-In-First-Out (FIFO) principle. However, Python offers variations such as LIFO (Last-In-First-Out) queues and priority queues. Here's an overview of each type:
queue.QueueDescription: A thread-safe FIFO queue that is ideal for multi-threaded programming. It ensures that the order of tasks or data is preserved, and it handles thread synchronization.
Basic Usage:
from queue import Queue
q = Queue()
q.put(1)
q.put(2)
q.put(3)
print(q.get()) # Output: 1
print(q.get()) # Output: 2
print(q.get()) # Output: 3
Thread Safety: The queue.Queue class includes built-in locking mechanisms that make it safe to use in multi-threaded applications.
Common Methods:
put(item, block=True, timeout=None): Adds an item to the queue. If the queue is full, it will block until space is available.get(block=True, timeout=None): Removes and returns an item from the queue. If the queue is empty, it will block until an item is available.task_done(): Indicates that a formerly enqueued task is complete.join(): Blocks until all items in the queue have been processed.Example: Implementing a Producer-Consumer Pattern:
import threading
from queue import Queue
def producer(queue):
for i in range(5):
queue.put(i)
print(f'Produced {i}')
def consumer(queue):
while not queue.empty():
item = queue.get()
print(f'Consumed {item}')
queue.task_done()
q = Queue()
producer_thread = threading.Thread(target=producer, args=(q,))
consumer_thread = threading.Thread(target=consumer, args=(q,))
producer_thread.start()
producer_thread.join()
consumer_thread.start()
consumer_thread.join()
q.join()
queue.LifoQueueDescription: A thread-safe LIFO queue, also known as a stack. The last item added is the first one to be removed.
Basic Usage:
from queue import LifoQueue
q = LifoQueue()
q.put(1)
q.put(2)
q.put(3)
print(q.get()) # Output: 3
print(q.get()) # Output: 2
print(q.get()) # Output: 1
Thread Safety: Similar to queue.Queue, LifoQueue is thread-safe and can be used in multi-threaded applications.
Common Methods:
put(item, block=True, timeout=None): Adds an item to the queue. If the queue is full, it will block until space is available.get(block=True, timeout=None): Removes and returns an item from the queue. If the queue is empty, it will block until an item is available.Example: Implementing a LIFO Stack in a Multi-Threaded Context:
import threading
from queue import LifoQueue
def push_items(queue):
for i in range(5):
queue.put(i)
print(f'Pushed {i}')
def pop_items(queue):
while not queue.empty():
item = queue.get()
print(f'Popped {item}')
q = LifoQueue()
push_thread = threading.Thread(target=push_items, args=(q,))
pop_thread = threading.Thread(target=pop_items, args=(q,))
push_thread.start()
push_thread.join()
pop_thread.start()
pop_thread.join()
queue.PriorityQueueDescription: A thread-safe priority queue where elements are retrieved in order based on their priority. Lower priority numbers are retrieved first.
Basic Usage:
from queue import PriorityQueue
q = PriorityQueue()
q.put((2, "second"))
q.put((1, "first"))
q.put((3, "third"))
print(q.get()) # Output: (1, "first")
print(q.get()) # Output: (2, "second")
print(q.get()) # Output: (3, "third")
Thread Safety: Like queue.Queue and LifoQueue, PriorityQueue is thread-safe.
Common Methods:
put(item, block=True, timeout=None): Adds an item to the queue based on its priority.get(block=True, timeout=None): Removes and returns the item with the highest priority (lowest priority number).Example: Task Scheduling with PriorityQueue:
import threading
from queue import PriorityQueue
def schedule_tasks(queue):
tasks = [(2, "second"), (1, "first"), (3, "third")]
for task in tasks:
queue.put(task)
print(f'Scheduled: {task}')
def execute_tasks(queue):
while not queue.empty():
task = queue.get()
print(f'Executing: {task}')
queue.task_done()
q = PriorityQueue()
schedule_thread = threading.Thread(target=schedule_tasks, args=(q,))
execute_thread = threading.Thread(target=execute_tasks, args=(q,))
schedule_thread.start()
schedule_thread.join()
execute_thread.start()
execute_thread.join()
q.join()
heapqDescription: A non-thread-safe priority queue implemented as a binary heap. This module provides heap operations like heappush and heappop.
Basic Usage:
import heapq
heap = []
heapq.heappush(heap, (2, "second"))
heapq.heappush(heap, (1, "first"))
heapq.heappush(heap, (3, "third"))
print(heapq.heappop(heap)) # Output: (1, "first")
print(heapq.heappop(heap)) # Output: (2, "second")
print(heapq.heappop(heap)) # Output: (3, "third")
Thread Safety: heapq is not thread-safe, so it should not be used in multi-threaded environments without external synchronization.
Common Methods:
heappush(heap, item): Pushes an item onto the heap, maintaining the heap property.heappop(heap): Pops and returns the smallest item from the heap.heapify(x): Converts a list into a heap, in-place, in linear time.Example: Implementing a Priority Queue with heapq:
import heapq
def push_items(heap):
items = [(3, "third"), (1, "first"), (2, "second")]
for item in items:
heapq.heappush(heap, item)
print(f'Pushed: {item}')
def pop_items(heap):
while heap:
item = heapq.heappop(heap)
print(f'Popped: {item}')
heap = []
push_items(heap)
pop_items(heap)
collections.dequeDescription: A double-ended queue that supports appending and popping from both ends with O(1) time complexity. deque is not thread-safe by default.
Basic Usage:
from collections import deque
d = deque()
d.append(1)
d.append(2)
d.appendleft(0)
print(d.pop()) # Output: 2
print(d.popleft()) # Output: 0
Thread Safety: deque is not thread-safe, so it should not be used in multi-threaded environments without external synchronization.
Common Methods:
append(x): Adds x to the right end of the deque.appendleft(x): Adds x to the left end of the deque.pop(): Removes and returns the rightmost element.popleft(): Removes and returns the leftmost element.extend(iterable): Extends the right end of the deque by appending elements from the iterable.extendleft(iterable): Extends the left end of the deque by appending elements from the iterable (note that the iterable is reversed).rotate(n): Rotates the deque n steps to the right. If n is negative, it rotates to the left.Example: Implementing a Double-Ended Queue with deque:
from collections import deque
d = deque(['a', 'b', 'c'])
d.append('d')
d.appendleft('z')
print(d) # Output: deque(['z', 'a', 'b', 'c', 'd'])
d.pop()
print(d) # Output: deque(['z', 'a', 'b', 'c'])
d.popleft()
print(d) # Output: deque(['a', 'b', 'c'])
| Feature/Queue | queue.Queue |
queue.LifoQueue |
queue.PriorityQueue |
heapq |
collections.deque |
|---|---|---|---|---|---|
| Thread-Safe | Yes | Yes | Yes | No | No |
| Time Complexity (Enqueue/Dequeue) | O(1) | O(1) | O(log n) | O(log n) | O(1) |
| Order | FIFO | LIFO | Priority (min-heap) | Priority (min-heap) | FIFO/LIFO (Both ends) |
| Ideal Use Case | Multi-threaded FIFO queue | Multi-threaded LIFO queue | Multi-threaded priority queue | Single-threaded priority queue | Fast append/pop from both ends |
| Memory Efficiency | Moderate | Moderate | Less efficient (due to heap) | Moderate | High |
| Extra Features | Block until available | Block until available | Block until available | No blocking | No blocking |
This comprehensive guide should help you understand the different queue types available in Python, their features, and the best practices for using them effectively in your programs.