Python 基于 threading 和 Queue 实现线程池的完整指南

沈知远

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

原创

本文详解如何利用Python标准库中的threading和Queue模块手动构建线程池。通过解析线程池存在的必要性、Queue核心API以及具体的代码实现步骤，帮助开发者掌握高并发场景下的资源管理与任务调度技巧，包含动态停止线程的高级用法。

线程池的必要性

一、线程池

1、为什么需要使用线程池

1.1 创建和销毁线程会引入系统开销，若频繁进行线程的创建与销毁，将显著降低处理效率。

假设创建线程耗时为T1，执行任务耗时为T2，销毁线程耗时为T3。如果T1+T3大于T2，则说明每次新建线程来执行单一任务是不经济的。线程池通过复用已有的空闲线程来执行新任务，有效避免了因频繁创建和销毁线程带来的系统开销。

1.2 线程并发数量过多，可能抢占系统资源从而导致阻塞。

线程共享系统资源，如果同时执行的线程数量过多，可能导致系统资源不足，进而引发阻塞情况。

1.3 对线程进行一些简单的管理。

例如：延时执行、定时循环执行的策略等，运用线程池都能进行很好的实现。

Python构建线程池的方法

2、Python中建立线程池的方法

2.1 使用threadpool模块，这是个python的第三方模块，支持python2和python3

2.2 使用concurrent.futures模块，这个模块是python3中自带的模块，python2.7以上版本也可以安装使用

2.3 自己构建一个线程池

Queue队列核心机制

二、队列（queue）

Queue模块提供的队列（FIFO）适用于多线程编程，在生产者(producer)和消费者(consumer)之间线程安全(thread-safe)地传递消息或其它数据，因此多个线程可以共用同一个Queue实例。常用方法:

Queue.qsize()：返回queue的大小。

Queue.empty():判断队列是否为空，通常不太靠谱。

Queue.full():判断是否满了。

Queue.put(item, block=True, timeout=None): 往队列里放数据。

Queue.put_nowait(item):往队列里存放元素，不等待

Queue.get(item, block=True, timeout=None): 从队列里取数据。

Queue.get_nowait(item):从队列里取元素，不等待

Queue.task_done()：表示队列中某个元素是否的使用情况，使用结束会发送信息。

Queue.join()：一直阻塞直到队列中的所有元素都执行完毕。

threading与Queue实现线程池实战

三、使用threading+Queue处理多任务

假设有十个任务需要处理，打算在后台开启五个线程，简化后的模型


import Queue
import threading
import time
 
queue = Queue.Queue()
 
class ThreadNum(threading.Thread):
  def __init__(self, queue):
    threading.Thread.__init__(self)
    self.queue = queue
 
  def run(self):
    while True:
      #消费者端，从队列中获取num
      num = self.queue.get()
      print("Retrieved", num)
      time.sleep(1) 
      #在完成这项工作之后，使用 queue.task_done() 函数向任务已经完成的队列发送一个信号
      self.queue.task_done()
    
    print("Consumer Finished")
 
def main():
  #产生一个 threads pool, 并把消息传递给thread函数进行处理，这里开启10个并发
  for i in range(5):
    t = ThreadNum(queue)
    t.setDaemon(True)
    t.start()
  
  #往队列中填数据 
  for num in range(10):
    queue.put(num)
    #wait on the queue until everything has been processed
  
  queue.join()
   
if __name__ == '__main__':
  main()
  time.sleep(500)

输出为：


('Retrieved', 0)
 ('Retrieved', 1)('Retrieved', 2)
('Retrieved', 3)
('Retrieved', 4)
('Retrieved', 5)('Retrieved', 6)
('Retrieved', 7)
('Retrieved', 8)
 ('Retrieved', 9)

具体工作步骤描述如下：

1、创建一个 Queue.Queue() 的实例，然后使用数据对它进行填充。

2、将经过填充数据的实例传递给线程类，后者是通过继承 threading.Thread 的方式创建的。

3、生成守护线程池。

4、每次从队列中取出一个项目，并使用该线程中的数据和 run 方法以执行相应的工作。

5、在完成这项工作之后，使用 queue.task_done() 函数向任务已经完成的队列发送一个信号。

6、对队列执行 join 操作，实际上意味着等到队列为空，再退出主程序。

在使用这个模式时需要注意一点：通过将守护线程设置为 true，程序运行完自动退出。好处是在退出之前，可以对队列执行 join 操作、或者等到队列为空。

注意运行main函数后继续执行time.sleep(500)，可以观察到主线程未结束的情况下ThreadNum(queue)生成的线程还在运行。如果需要停止线程的话可以对以上代码加以修改。


import Queue
import threading
import time
 
queue = Queue.Queue()
 
class ThreadNum(threading.Thread):
  """没打印一个数字等待1秒，并发打印10个数字需要多少秒？"""
  def __init__(self, queue):
    threading.Thread.__init__(self)
    self.queue = queue
 
  def run(self):
    done = False
    while not done:
      #消费者端，从队列中获取num
      num = self.queue.get()
      if num is None:
        done = True
      else:
        print("Retrieved", num)
      time.sleep(1) 
      #在完成这项工作之后，使用 queue.task_done() 函数向任务已经完成的队列发送一个信号
      self.queue.task_done()
    
    print("Consumer Finished")
def main():
  #产生一个 threads pool, 并把消息传递给thread函数进行处理，这里开启10个并发
  for i in range(5):
    t = ThreadNum(queue)
    t.setDaemon(True)
    t.start()
  
  #往队列中填错数据 
  for num in range(10):
    queue.put(num)
  
  queue.join()
  time.sleep(100)
  for i in range(10):
    queue.put(None)
    print('None')
  time.sleep(200)
   
if __name__ == '__main__':
  start = time.time()
  main()
  print"Elapsed Time: %s" % (time.time() - start)

main函数执行完后队列向线程发送None消息，触发线程的停止标识，这样就可以动态管理线程池了。

以上这篇Python 使用threading+Queue实现线程池示例就是小编分享给大家的全部内容了，希望能给大家一个参考，也希望大家多多支持本站。