当前位置：首页 > news >正文

Python的`queue`模块

news 来源：原创 2024/9/20 7:41:50

队列（Queue）

在Python的queue模块中，Queue类是一个线程安全的队列实现，用于在多线程编程中安全地交换信息。它遵循先入先出（FIFO）的原则。Queue类提供了几种主要的方法：

put(item): 将一个项目放入队列中。如果队列已满，则此方法将阻塞，直到队列中有空间可用。
get(): 从队列中移除并返回一个项目。如果队列为空，则此方法将阻塞，直到队列中有项目可用。
task_done(): 表示之前入队的一个任务已经完成。通常由使用get()的消费者线程调用。每调用一次get()，随后就应该调用一次task_done()来告诉队列该任务已处理完毕。
join(): 阻塞调用线程，直到队列中的所有项目都被处理并调用task_done()方法。这通常用于等待由队列消费的所有工作项完成。

并发编程中的线程

在Python中，threading模块提供了基本的线程和锁支持。每个Thread对象代表了一个线程的执行。线程之间共享全局解释器锁（GIL），这意味着在任意时刻只有一个线程可以执行Python字节码。然而，对于I/O密集型任务（如文件读写、网络请求）和等待密集型任务（如数据库查询、外部API调用），多线程仍然可以提高程序的总体性能，因为线程在等待时可以释放GIL，允许其他线程运行。

示例详解

下面是对之前示例的详细解释，包括一些注意事项和潜在的改进点：

import queue
import threading
import time
import randomdef worker(q, job_type):while True:  # 注意：这里的无限循环需要外部机制来中断if job_type == 'producer':item = random.randint(1, 100)q.put(item)print(f'{threading.current_thread().name} 生产了 {item}')time.sleep(random.random())  # 模拟耗时操作elif job_type == 'consumer':item = q.get()print(f'{threading.current_thread().name} 消费了 {item}')time.sleep(random.random())  # 模拟耗时操作q.task_done()  # 告诉队列这个任务已完成# 创建一个队列
q = queue.Queue(maxsize=10)  # 可选地设置队列的最大大小# 创建并启动生产者线程
producers = [threading.Thread(target=worker, args=(q, 'producer'), name=f'生产者-{i+1}') for i in range(2)]
for t in producers:t.daemon = Truet.start()# 创建并启动消费者线程
consumers = [threading.Thread(target=worker, args=(q, 'consumer'), name=f'消费者-{i+1}') for i in range(3)]
for t in consumers:t.daemon = Truet.start()# 注意：由于生产者线程是无限循环的，下面的q.join()将永远不会返回
# 除非有某种机制来停止生产者线程（例如，使用共享变量或threading.Event）
# 为了示例的简洁性，这里省略了停止生产者的代码# 假设我们在某个时刻决定停止所有线程
# 这可以通过设置共享变量或使用threading.Event来实现
# ...（这里省略了停止线程的代码）# 主线程通常不会在这里等待，因为它会立即继续执行后面的代码
# 但为了示例的完整性，我们在这里添加了一个简单的阻塞，以便可以看到一些输出
try:while True:time.sleep(1)
except KeyboardInterrupt:print("主线程被中断，尝试优雅地关闭所有线程...")# 在这里添加关闭所有线程的代码（如果有的话）# 注意：由于我们设置了daemon=True，主线程结束时守护线程也会自动结束# 但这通常不是优雅关闭线程的好方法