Implement a FIFO Message Queue in Python with deque
This code implements a FIFO (first-in-first-out) message queue class using Python's collections.deque, providing enqueue, dequeue, peek, and size operations.
Python code
44 linesfrom collections import deque
class MessageQueue:
def __init__(self):
self.queue = deque()
def enqueue(self, message):
self.queue.append(message)
print(f"Enqueued: {message}")
def dequeue(self):
if self.is_empty():
print("Queue is empty, cannot dequeue.")
return None
message = self.queue.popleft()
print(f"Dequeued: {message}")
return message
def peek(self):
if self.is_empty():
print("Queue is empty, nothing to peek.")
return None
message = self.queue[0]
print(f"Front of queue: {message}")
return message
def is_empty(self):
return len(self.queue) == 0
def size(self):
print(f"Queue size: {len(self.queue)}")
return len(self.queue)
if __name__ == "__main__":
mq = MessageQueue()
mq.enqueue("Message 1")
mq.enqueue("Message 2")
mq.enqueue("Message 3")
mq.peek()
mq.size()
mq.dequeue()
mq.dequeue()
mq.dequeue()
mq.dequeue()
Output
Enqueued: Message 1
Enqueued: Message 2
Enqueued: Message 3
Front of queue: Message 1
Queue size: 3
Dequeued: Message 1
Dequeued: Message 2
Dequeued: Message 3
Queue is empty, cannot dequeue.
How it works
The deque provides O(1) append and popleft operations, making it an efficient structure for a FIFO queue as opposed to a list where popping from the front is O(n). The enqueue method appends to the right, while dequeue removes from the left, preserving order. The peek method returns the front element without removal, and is_empty checks the queue's length. This class encapsulates the queue logic, allowing easy reuse in larger applications.
Common mistakes
- Using a list and pop(0) which is O(n) instead of deque.popleft()
- Not handling dequeue/peek on an empty queue, leading to IndexError
- Forgetting to check the queue size before dequeue in concurrent scenarios
- Using appendleft for enqueue which reverses the FIFO order
Variations
- Use queue.Queue from the standard library for thread-safe FIFO operations
- Implement with a list and an index pointer to avoid shifting elements
Real-world use cases
- Managing a task queue in a multi-threaded application to process jobs in order.
- Buffering messages in a microservice before batch processing for ordered delivery.
- Implementing a request queue for an API gateway to balance load sequentially.
Sponsored
More from Streaming & messaging
- At Most Once Fire-and-Forget Mock in Python easy
- Batch Consume Process Commit Pattern in Python medium
- Build a Streaming Messaging Helper in Python easy
- Dead Letter Queue Failed Messages List Mock in Python easy
- Dedupe processed message IDs in Python easy
- Event Envelope with Schema Version Field in Python easy
Keep learning
Related tutorials and quizzes for this topic.