Python Queue模块详解：三种队列及多线程应用实例

小石头

发布时间：2026-10-13| 1人浏览过

原创

本文详细介绍Python中Queue模块的使用方法，涵盖FIFO、LIFO及优先级队列三种类型。通过解析put()与get()方法的阻塞机制，结合多线程生成随机数并筛选奇偶数的完整代码示例，帮助开发者掌握线程间数据交换的最佳实践。

Python Queue模块

在Python中，队列是线程间交换数据最常用的方式。Queue模块提供了队列操作功能，尽管其接口简洁易用，但若使用不当，仍可能引发一些意外情况。

创建一个“队列”对象

import Queue
q = Queue.Queue(maxsize = 10)

Queue.Queue类实现了同步队列，其长度可以设置为无限或有限。通过构造函数的可选参数maxsize可以指定队列长度；若maxsize小于1，则代表队列长度无限制。

将一个值放入队列中

q.put(10)

调用队列对象的put()方法可在队尾插入新项。put()包含两个参数：第一个item为必填项，即要插入的值；第二个block为可选参数，默认值为1。当队列当前为空且block为1时，put()会使调用线程暂停，直到队列中出现可用空间。若block设为0，当队列满时put()将抛出Full异常。

将一个值从队列中取出

q.get()

调用队列对象的get()方法可从队头删除并返回一个项目。该方法的可选参数为block，默认值为True。若队列为空且block为True，get()会使调用线程暂停，直到有项目可用。若队列为空且block为False，get()将抛出Empty异常。

Python Queue模块提供三种队列及其构造函数：

1、FIFO队列（先进先出）：class Queue.Queue(maxsize) 2、LIFO队列（类似堆栈，先进后出）：class Queue.LifoQueue(maxsize) 3、优先级队列（级别越低越先取出）：class Queue.PriorityQueue(maxsize)

此包中常用的方法（假设q = Queue.Queue()）：

q.qsize() 返回队列的大小
q.empty() 如果队列为空，返回True,反之False
q.full() 如果队列满了，返回True,反之False
q.full 与 maxsize 大小对应
q.get([block[, timeout]]) 获取队列，timeout等待时间
q.get_nowait() 相当q.get(False)
非阻塞 q.put(item) 写入队列，timeout等待时间
q.put_nowait(item) 相当q.put(item, False)
q.task_done() 在完成一项工作之后，q.task_done() 函数向任务已经完成的队列发送一个信号
q.join() 实际上意味着等到队列为空，再执行别的操作

范例：

实现一个线程不断生成随机数并存入队列（建议使用Queue模块）。

实现另一个线程从上述队列中不断取出奇数。

实现第三个线程从上述队列中不断取出偶数。

#!/usr/bin/env python
#coding:utf8
import random,threading,time
from Queue import Queue
#Producer thread
class Producer(threading.Thread):
  def __init__(self, t_name, queue):
    threading.Thread.__init__(self,name=t_name)
    self.data=queue
  def run(self):
    for i in range(10):  #随机产生10个数字 ，可以修改为任意大小
      randomnum=random.randint(1,99)
      print "%s: %s is producing %d to the queue!" % (time.ctime(), self.getName(), randomnum)
      self.data.put(randomnum) #将数据依次存入队列
      time.sleep(1)
    print "%s: %s finished!" %(time.ctime(), self.getName())
 
#Consumer thread
class Consumer_even(threading.Thread):
  def __init__(self,t_name,queue):
    threading.Thread.__init__(self,name=t_name)
    self.data=queue
  def run(self):
    while 1:
      try:
        val_even = self.data.get(1,5) #get(self, block=True, timeout=None) ,1就是阻塞等待,5是超时5秒
        if val_even%2==0:
          print "%s: %s is consuming. %d in the queue is consumed!" % (time.ctime(),self.getName(),val_even)
          time.sleep(2)
        else:
          self.data.put(val_even)
          time.sleep(2)
      except:   #等待输入，超过5秒 就报异常
        print "%s: %s finished!" %(time.ctime(),self.getName())
        break
class Consumer_odd(threading.Thread):
  def __init__(self,t_name,queue):
    threading.Thread.__init__(self, name=t_name)
    self.data=queue
  def run(self):
    while 1:
      try:
        val_odd = self.data.get(1,5)
        if val_odd%2!=0:
          print "%s: %s is consuming. %d in the queue is consumed!" % (time.ctime(), self.getName(), val_odd)
          time.sleep(2)
        else:
          self.data.put(val_odd)
          time.sleep(2)
      except:
        print "%s: %s finished!" % (time.ctime(), self.getName())
        break
#Main thread
def main():
  queue = Queue()
  producer = Producer('Pro.', queue)
  consumer_even = Consumer_even('Con_even.', queue)
  consumer_odd = Consumer_odd('Con_odd.',queue)
  producer.start()
  consumer_even.start()
  consumer_odd.start()
  producer.join()
  consumer_even.join()
  consumer_odd.join()
  print 'All threads terminate!'
 
if __name__ == '__main__':
  main()