如何在Python中實施用於並發編程的線程安全隊列？

使用queue.Queue是實現Python線程安全隊列的最可靠方法，1. 它內置鎖機制，避免競態條件；2. put()和get()默認阻塞，支持timeout避免無限等待；3. 配合task_done()和join()可協調任務完成；4. 可設置maxsize實現有界隊列以控制內存；5. 自定義封裝可提供更清晰接口；6. 注意queue.Queue僅用於線程間通信，不適用於多進程場景，且若使用join()則必須在消費者中調用task_done()，否則程序可能掛起。

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.

以上是如何在Python中實施用於並發編程的線程安全隊列？的詳細內容。更多資訊請關注PHP中文網其他相關文章！