How to implement a thread-safe queue for concurrent programming in Python?

Use queue.Queue is the most reliable method to implement Python thread-safe queues. 1. It has a built-in lock mechanism to avoid race conditions; 2. Put() and get() block by default, and support timeout to avoid infinite waiting; 3. Use task_done() and join() to coordinate task completion; 4. You can set maxsize to implement bounded queues to control memory; 5. Custom encapsulation can provide a clearer interface; 6. Note that queue.Queue is only used for inter-thread communication and is not suitable for multi-process scenarios. If you use join(), task_done() must be called in the consumer, otherwise the program may hang.

Implementing a thread-safe queue in Python is straightforward thanks to the built-in queue.Queue class, which is designed specifically for safe use across multiple threads. Here's how to do it properly and what to keep in mind.

Use queue.Queue for Thread Safety

The most reliable and simplest way to have a thread-safe queue in Python is to use the queue.Queue module. It handles all locking mechanisms internally, so you don't have to worry about race conditions when putting or getting items.

 import queue
import threading
import time

# Create a thread-safe queue
q = queue.Queue()

def producer():
    for i in range(5):
        item = f"item-{i}"
        q.put(item)
        print(f"Produced: {item}")
        time.sleep(0.1) # Simulate work

def consumer():
    While True:
        try:
            # Use timeout to avoid blocking forever
            item = q.get(timeout=1)
            print(f"Consumed: {item}")
            q.task_done() # Indicate that the task is done
        except queue.Empty:
            print("No more items, exiting.")
            break

# Create threads
t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)

# Start threads
t1.start()
t2.start()

# Wait for both threads to finish
t1.join()
t2.join()

Key Points for Using queue.Queue

• Blocking Operations : put() and get() block by default, which is useful for coordinating threads.
• task_done() and join() : If you use task_done() after processing each item, you can call q.join() in the producer to wait until all items are processed.
• Avoid queue.Empty and queue.Full exceptions : Always handle them when using non-blocking or timeout-based calls.

Example with q.join() :

 def producer_with_join():
    for i in range(3):
        q.put(f"task-{i}")
    q.join() # Wait until all tasks are marked as done
    print("All tasks completed.")

def consumer_with_task_done():
    While True:
        item = q.get()
        if item is None:
            break
        print(f"Processing {item}")
        time.sleep(0.5)
        q.task_done() # Mark task as done

Alternative: Bounded Queue (With Size Limit)

You can limit the queue size to control memory usage and enable backpressure:

 q = queue.Queue(maxsize=3)

Now, put() will block if the queue is full until an item is consumed.

Custom Thread-Safe Queue (Advanced Use Case)

While queue.Queue covers most needs, you might want to wrap it for specific behavior:

 import queue
from typing import Any

class SafeQueue:
    def __init__(self, maxsize: int = 0):
        self._q = queue.Queue(maxsize=maxsize)

    def put(self, item: Any):
        self._q.put(item)

    def get(self) -> Any:
        return self._q.get()

    def empty(self) -> bool:
        return self._q.empty()

    def size(self) -> int:
        return self._q.qsize()

    def task_done(self):
        self._q.task_done()

    def join(self):
        self._q.join()

This encapsulates the queue and provides a clean interface.

Important Notes

• Don't use list with locks manually unless you have a very specific reason — it's error-prone.
• queue.Queue is for threads only. For multiprocessing, use multiprocessing.Queue .
• Always call task_done() in consumers if you use join() , or your program may hang.

Basically, just use queue.Queue — it's well-tested, efficient, and built for this exact purpose.

The above is the detailed content of How to implement a thread-safe queue for concurrent programming in Python?. For more information, please follow other related articles on the PHP Chinese website!