Python实现线程池

13年前

最近在做一些文本处理方面的事情，考虑到程序利用并发性可以提高执行效率（不纠结特殊反例），于是入围的Idea如使用多进程或多线程达到期望的目标，对于进程或线程的创建是有代价的，那么我们是否可以实现一个线程池来达到已创建的线程反复使用从而使代价降低到最小呢？
当然可以，要想创建一个线程池，那么必须得有个容器来模拟“池”，在Python中，队列这样的数据结构就可以帮我们解决“池”这个问题，然而随之引来的多线程之间数据处理同步问题，好在Python中有个Queue模块帮我们解决了这一棘手的问题，那么我们就可以实现一个线程池的雏形了。

Python代码
  1. # !/usr/bin/env python   
  2. # -*- coding:utf-8 -*-   
  3.   
  4. import Queue   
  5. import threading   
  6. import time   
  7.   
  8. class WorkManager(object):   
  9.     def __init__(self, work_num=1000,thread_num=2):   
  10.         self.work_queue = Queue.Queue()   
  11.         self.threads = []   
  12.         self.__init_work_queue(work_num)   
  13.         self.__init_thread_pool(thread_num)   
  14.   
  15.     """  
  16.         初始化线程  
  17.     """  
  18.     def __init_thread_pool(self,thread_num):   
  19.         for i in range(thread_num):   
  20.             self.threads.append(Work(self.work_queue))   
  21.   
  22.     """  
  23.         初始化工作队列  
  24.     """  
  25.     def __init_work_queue(self, jobs_num):   
  26.         for i in range(jobs_num):   
  27.             self.add_job(do_job, i)   
  28.   
  29.     """  
  30.         添加一项工作入队  
  31.     """  
  32.     def add_job(self, func, *args):   
  33.         self.work_queue.put((func, list(args)))#任务入队，Queue内部实现了同步机制   
  34.   
  35.     """  
  36.         等待所有线程运行完毕  
  37.     """      
  38.     def wait_allcomplete(self):   
  39.         for item in self.threads:   
  40.             if item.isAlive():item.join()   
  41.   
  42. class Work(threading.Thread):   
  43.     def __init__(self, work_queue):   
  44.         threading.Thread.__init__(self)   
  45.         self.work_queue = work_queue   
  46.         self.start()   
  47.   
  48.     def run(self):   
  49.         #死循环，从而让创建的线程在一定条件下关闭退出   
  50.         while True:   
  51.             try:   
  52.                 do, args = self.work_queue.get(block=False)#任务异步出队，Queue内部实现了同步机制   
  53.                 do(args)   
  54.                 self.work_queue.task_done()#通知系统任务完成   
  55.             except:   
  56.                 break  
  57.   
  58. #具体要做的任务   
  59. def do_job(args):   
  60.     time.sleep(0.1)#模拟处理时间   
  61.     print threading.current_thread(), list(args)   
  62.   
  63. if __name__ == '__main__':   
  64.     start = time.time()   
  65.     work_manager =  WorkManager(10000, 10)#或者work_manager =  WorkManager(10000, 20)   
  66.     work_manager.wait_allcomplete()   
  67.     end = time.time()   
  68.     print "cost all time: %s" % (end-start)  

 2次开启不同的线程数运行结果如下：

Python代码
  1. #work_manager =  WorkManager(10000, 10)   
  2. cost all time: 100.641790867(单位：秒)   
  3. #work_manager =  WorkManager(10000, 20)   
  4. cost all time：50.5233478546(单位：秒)  

 上面实现了线程池的雏形，展现了基本原理，当然要想成为通用的API需要做很多的工作，希望本文能够起到抛砖引玉的效果。