Thread-Safe Producer Consumer Queue in Python
A producer-consumer pattern using thread-safe queue.Queue with two threads, demonstrating safe communication and synchronized task completion.
Python code
39 linesimport queue
import threading
import time
import random
def producer(q, item_count):
for i in range(item_count):
item = random.randint(1, 100)
q.put(item)
print(f"Producer added: {item}")
time.sleep(0.1)
def consumer(q):
while True:
try:
item = q.get(timeout=2)
print(f"Consumer got: {item}")
q.task_done()
except queue.Empty:
print("Queue empty, consumer exiting.")
break
if __name__ == "__main__":
q = queue.Queue(maxsize=5)
item_count = 5
producer_thread = threading.Thread(target=producer, args=(q, item_count))
consumer_thread = threading.Thread(target=consumer, args=(q,))
producer_thread.start()
consumer_thread.start()
producer_thread.join()
q.join()
consumer_thread.join()
print("All done.")
Output
Producer added: 42
Consumer got: 42
Producer added: 17
Consumer got: 17
Producer added: 93
Consumer got: 93
Producer added: 55
Consumer got: 55
Producer added: 82
Consumer got: 82
Queue empty, consumer exiting.
All done.
How it works
The queue.Queue class is inherently thread-safe, so producers and consumers can add and remove items without explicit locks. q.task_done() signals that an item has been processed, and q.join() blocks until all items are marked done. The consumer uses q.get(timeout=2) to prevent infinite blocking, exiting with queue.Empty when no new items arrive. Producer and consumer threads run concurrently, demonstrating safe coordination through the queue.
Common mistakes
- Calling q.join() before all producers finish adding items
- Forgetting to call q.task_done() after processing each item
- Using a plain list instead of queue.Queue, which is not thread-safe
Variations
- Use a sentinel value like None to signal the consumer to stop, instead of relying on timeout.
- Switch to multiprocessing.Queue if you need parallelism across CPU cores.
Real-world use cases
- Log processors that enqueue incoming log lines and have worker threads write them to disk.
- Web scrapers that queue URLs for multiple fetcher threads to download pages concurrently.
- Image thumbnail generators that push image paths to a queue for background processing workers.
Sponsored
More from Concurrency & performance
- Benchmark list.append vs deque.append in Python medium
- Build a Python Performance Profiler That Generates Readable Reports medium
- Graceful Shutdown Executor Context Manager in Python medium
- How to Build a Producer-Consumer Pattern with asyncio.Queue in Python medium
- How to Cancel an asyncio Task with Graceful Cleanup in Python medium
- How to Convert Data in Parallel with ThreadPoolExecutor in Python easy
Keep learning
Related tutorials and quizzes for this topic.