Python多线程间的数据共享：队列的正确使用姿势

花韻仙語

发布时间：2025-08-13 20:04:33

997人浏览过

来源于php中文网

原创

python多线程间的数据共享：队列的正确使用姿势

本文旨在解决Python多线程环境下安全高效的数据共享问题。针对多个线程需要同时访问和处理同一份数据的情况，详细介绍了使用queue.Queue实现线程间数据传递的最佳实践。通过示例代码，展示了如何创建多个消费者队列，并由生产者线程将数据分发到所有队列，确保每个消费者都能独立地处理数据，避免数据竞争和线程阻塞。

在多线程Python程序中，线程间的数据共享是一个常见的需求。例如，一个线程负责从网络接收数据，而其他多个线程则负责处理这些数据。直接使用全局变量可能会导致线程安全问题，而使用单个队列则可能导致数据被一个线程取走后，其他线程无法访问。本文将介绍一种使用多个队列来实现数据共享的方法，确保每个线程都能独立地处理数据。

多队列方案：确保每个线程都能访问数据

针对多个消费者线程需要同时处理同一份数据的情况，最佳实践是为每个消费者线程创建一个独立的队列。生产者线程将数据复制到每个队列中，这样每个消费者线程都可以从自己的队列中获取数据进行处理，互不影响。

实现步骤

创建多个队列： 为每个消费者线程创建一个queue.Queue对象。
生产者线程： 生产者线程获取到数据后，将数据分别放入每个队列中。
消费者线程： 每个消费者线程从自己的队列中获取数据进行处理。
结束信号： 当生产者线程完成数据生产后，向每个队列发送一个结束信号（例如None），通知消费者线程退出。

示例代码

import threading
import time
from queue import Queue


def publisher(consumers):
    """
    生产者线程，负责将数据放入每个消费者队列中。
    """
    for x in range(10):
        value = 2 ** x
        for consumer in consumers:
            consumer.put(value)
        time.sleep(0.1)
    for consumer in consumers:
        consumer.put(None)  # sentinel value to indicate end of stream


def consumer(name, queue):
    """
    消费者线程，负责从自己的队列中获取数据并处理。
    """
    while True:
        value = queue.get()
        if value is None:
            print(f"{name} will quit now")
            break
        print(f"{name}: Got {value}")


def main():
    """
    主函数，创建并启动生产者和消费者线程。
    """
    consumer_threads = []
    consumer_queues = []
    for x in range(3):
        queue = Queue()
        consumer_queues.append(queue)
        thread = threading.Thread(target=consumer, args=(f"Consumer {x}", queue))
        thread.start()
        consumer_threads.append(thread)
    publisher_thread = threading.Thread(target=publisher, args=(consumer_queues,))
    publisher_thread.start()
    publisher_thread.join()  # 等待生产者线程结束
    for thread in consumer_threads:
        thread.join()  # 等待所有消费者线程结束


if __name__ == "__main__":
    main()

代码解释

publisher(consumers) 函数模拟生产者线程，它循环生成数据，并将数据放入每个消费者队列 consumers 中。
consumer(name, queue) 函数模拟消费者线程，它从自己的队列 queue 中获取数据并打印。当从队列中获取到 None 时，表示数据流结束，线程退出。
main() 函数创建了多个消费者队列和线程，并将它们传递给生产者线程。 publisher_thread.join() 和 thread.join() 确保主线程等待所有子线程完成后再退出。