Python3教程：queue 模块用法-猿码集

Python3教程：queue 模块用法

1. 引言

当我们需要在一个线程中将数据传递给另一个线程时，可以使用多种方式实现，其中一种方式是使用队列 (queue)。在 Python 中，自带一个 queue 模块，提供了 FIFO 队列 (Queue)、LIFO 队列 (LifoQueue)、优先队列 (PriorityQueue) 等多种队列。本文将重点介绍 Python 中 queue 模块的使用，并且会提供一些示例代码，帮助读者更好地理解。

2. 基本操作

2.1 创建队列

首先需要导入 queue 模块，然后就可以创建队列了。创建队列时，需要指定队列的类型，包括 FIFO、LIFO 和优先队列。FIFO 队列是一种先进先出的队列，LIFO 队列是一种后进先出的队列，而优先队列则是基于优先级进行排序的队列。以下是使用 queue 模块创建队列的基本步骤：


import queue
# 创建 FIFO 队列
fifo_queue = queue.Queue()
# 创建 LIFO 队列
lifo_queue = queue.LifoQueue()
# 创建优先队列
priority_queue = queue.PriorityQueue()

2.2 放入数据

向队列中添加数据需要使用 put() 方法。put() 方法可以有一个可选的 timeout 参数，用于设置阻塞时间，如果队列满了，会等待 timeout 秒后再继续执行操作。


# 向队列中添加数据
fifo_queue.put("A")
fifo_queue.put("B")
fifo_queue.put("C")
lifo_queue.put("A")
lifo_queue.put("B")
lifo_queue.put("C")
priority_queue.put((3, "A"))  # 添加一个优先级为 3 的元素 "A"
priority_queue.put((1, "B"))  # 添加一个优先级为 1 的元素 "B"
priority_queue.put((2, "C"))  # 添加一个优先级为 2 的元素 "C"

2.3 获取数据

从队列中获取数据可以使用 get() 方法。get() 方法也可以有一个可选的 timeout 参数，用于设置阻塞时间，如果队列为空，会等待 timeout 秒后再继续执行操作。


# 获取队列中的数据
print(fifo_queue.get())  # 输出：A
print(fifo_queue.get())  # 输出：B
print(fifo_queue.get())  # 输出：C
print(lifo_queue.get())  # 输出：C
print(lifo_queue.get())  # 输出：B
print(lifo_queue.get())  # 输出：A
print(priority_queue.get())  # 输出：(1, 'B')
print(priority_queue.get())  # 输出：(2, 'C')
print(priority_queue.get())  # 输出：(3, 'A')

3. 队列的属性和方法

3.1 队列的大小

可以使用 qsize() 方法获取队列的大小。


# 获取队列的大小
print(fifo_queue.qsize())  # 输出：0
print(lifo_queue.qsize())  # 输出：0
print(priority_queue.qsize())  # 输出：0

3.2 判断队列是否为空

可以使用 empty() 方法判断队列是否为空。


# 判断队列是否为空
print(fifo_queue.empty())  # 输出：True
print(lifo_queue.empty())  # 输出：True
print(priority_queue.empty())  # 输出：True

3.3 判断队列是否已满

可以使用 full() 方法判断队列是否已满。注意，LIFO 队列不支持 full() 方法。


# 判断队列是否已满
print(fifo_queue.full())  # 输出：False
print(priority_queue.full())  # 输出：False

3.4 清空队列

可以使用 queue.clear() 方法清空队列。


# 清空队列
fifo_queue.clear()
lifo_queue.clear()
priority_queue.clear()

3.5 队列的阻塞操作

当队列为空时，调用 get() 方法会阻塞代码。当队列满时，调用 put() 方法也会阻塞代码。以下是一些常用的阻塞操作：

get(block=True, timeout=None)：当队列为空时，调用该方法会一直阻塞代码，直到有新的数据入队或超时。

put(item, block=True, timeout=None)：当队列已满时，调用该方法会一直阻塞代码，直到有数据出队或超时。

join()：等待队列中的所有任务执行完毕。

task_done()：通知队列已完成一个任务。

4. 示例代码

下面是一些示例代码，演示了如何使用 queue 模块。

4.1 生产者-消费者模型

生产者-消费者模型是一种常见的多线程模型。本示例演示了如何使用队列实现生产者-消费者模型。


import queue
import threading
import random
import time
class Producer(threading.Thread):
    def __init__(self, queue):
        threading.Thread.__init__(self)
        self.queue = queue
    def run(self):
        while True:
            # 随机生成一个数字
            item = random.randint(1, 100)
            # 放入队列中
            self.queue.put(item)
            # 打印提示信息
            print("[Producer] Added item: {} [queue size={}]".format(item, self.queue.qsize()))
            # 随机等待一段时间
            time.sleep(random.uniform(0.5, 1.5))
class Consumer(threading.Thread):
    def __init__(self, queue):
        threading.Thread.__init__(self)
        self.queue = queue
    def run(self):
        while True:
            # 获取数据
            item = self.queue.get()
            # 打印提示信息
            print("[Consumer] Removed item: {} [queue size={}]".format(item, self.queue.qsize()))
            # 通知任务已完成
            self.queue.task_done()
            # 随机等待一段时间
            time.sleep(random.uniform(0.5, 1.5))
if __name__ == "__main__":
    q = queue.Queue()
    # 启动生产者线程
    t1 = Producer(q)
    t1.start()
    # 启动消费者线程
    t2 = Consumer(q)
    t2.start()
    # 等待队列中的所有任务执行完毕
    q.join()

重点提示：在生产者-消费者模型中，生产者线程会不断地向队列中添加数据，而消费者线程会不断地取出队列中的数据，这两个操作会不断地交替进行。在代码中，每当生产者线程添加了一个数据时，都会打印一个提示信息。同样，在消费者线程取出一个数据时，也会打印一个提示信息。我们可以看到，即使在多线程的情况下，队列的使用也非常方便。

4.2 使用队列实现多线程下载

以下示例演示了如何使用队列实现多线程下载。在本例中，下载任务被分成若干个子任务，每个子任务由一个线程执行。每个线程都从队列中取出一个子任务进行下载，下载完成后继续从队列中取出下一个子任务。当队列为空时，表示所有任务已完成。


import os
import queue
import threading
import requests
class Downloader(threading.Thread):
    def __init__(self, queue, path):
        threading.Thread.__init__(self)
        self.queue = queue
        self.path = path
    def run(self):
        while True:
            # 从队列中获取一个子任务
            url = self.queue.get()
            # 组装数据的URL链接和本地文件的保存路径
            filename = os.path.join(self.path, url.split("/")[-1])
            # 下载数据
            response = requests.get(url, stream=True)
            with open(filename, "wb") as f:
                for chunk in response.iter_content(chunk_size=1024):
                    f.write(chunk)
            # 触发任务完成事件
            self.queue.task_done()
if __name__ == "__main__":
    url_list = [
        "https://cdn.pixabay.com/photo/2021/03/18/09/16/cat-6103309__340.jpg",
        "https://cdn.pixabay.com/photo/2015/04/23/22/00/tree-736885__340.jpg",
        "https://cdn.pixabay.com/photo/2021/03/15/13/53/cat-6091805__340.jpg",
        "https://cdn.pixabay.com/photo/2016/03/27/22/10/girl-1284451__340.jpg"
    ]
    # 创建队列
    q = queue.Queue()
    # 设置本地文件保存路径
    path = os.path.join(os.getcwd(), "downloads")
    if not os.path.exists(path):
        os.mkdir(path)
    # 创建下载任务线程
    for i in range(4):
        d = Downloader(q, path)
        d.setDaemon(True)
        d.start()
    # 将下载任务分解成小任务并放入队列中
    for url in url_list:
        q.put(url)
    # 等待所有任务完成
    q.join()
    print("所有下载任务已完成！")

重点提示：在多线程下载中，将下载任务分解成若干个子任务，并将子任务放入队列中。创建多个线程，在每个线程中不断从队列中取出一个子任务并执行下载操作。当队列为空时，表示所有任务已完成。