前面转载了一篇分析进程池源码的博文,是一篇分析进程池很全面的文章,。在Python中还有一个线程池的概念,它也有并发处理能力,在一定程度上能提高系统运行效率;不正之处欢迎批评指正。
线程的生命周期可以分为5个状态:创建、就绪、运行、阻塞和终止。自线程创建到终止,线程便不断在运行、创建和销毁这3个状态。一个线程的运行时间可由此可以分为3部分:线程的启动时间、线程体的运行时间和线程的销毁时间。在多线程处理的情景中,如果线程不能被重用,就意味着每次创建都需要经过启动、销毁和运行3个过程。这必然会增加系统相应的时间,降低了效率。看看之前介绍线程的博文的例子中(),有多少个任务,就创建多少个线程,但是由于Python特有的GIL限制,它并不是真正意义上的多线程,反而会因为频繁的切换任务等开销而降低了性能()。这种情况下可以使用线程池提高运行效率。
线程池的基本原理如下图,它是通过将事先创建多个能够执行任务的线程放入池中,所需要执行的任务通常要被安排在队列任务中。一般情况下,需要处理的任务比线程数目要多,线程执行完当前任务后,会从队列中取下一个任务,知道所有的任务完成。
由于线程预先被创建并放入线程池中,同时处理完当前任务之后并不销毁而是被安排处理下一个任务,因此能够避免多次创建线程,从而节省线程创建和销毁的开销,能带来更好的性能和系统稳定性。所以,说白了,Python的线程池也没有利用到多核或者多CPU的优势,只是跟普通的多线程相比,它不用去多次创建线程,节省了线程创建和销毁的时间,从而提高了性能。
Python中 线程池技术适合处理突发性大量请求或者需要大量线程来完成任务、但每个任务实际处理时间较短的场景,它能有效的避免由于系统创建线程过多而导致性能负荷过大、响应过慢等问题。下面介绍几种利用线程池的方法。
(一)自定义线程池模式
我们可以利用Queue模块和threading模块来实现线程池。Queue用来创建任务队列,threading用来创建一个线程池子。
看下面例子
import Queue,threadingclass Worker(threading.Thread): """ 定义一个能够处理任务的线程类,属于自定义线程类,自定义线程类就需要定义run()函数 """ def __init__(self,workqueue,resultqueue,**kwargs): threading.Thread.__init__(self,**kwargs) self.workqueue = workqueue#存放任务的队列,任务一般都是函数 self.resultqueue = resultqueue#存放结果的队列 def run(self): while True: try: #从任务队列中取出一个任务,block设置为False表示如果队列空了,就会抛出异常 callable,args,kwargs = self.workqueue.get(block=False) res = callable(*args,**kwargs) self.resultqueue.put(res)#将任务的结果存放到结果队列中 except Queue.Empty:#抛出空队列异常 breakclass WorkerManger(object): """ 定义一个线程池的类 """ def __init__(self,num=10):#默认这个池子里有10个线程 self.workqueue = Queue.Queue()#任务队列, self.resultqueue = Queue.Queue()#存放任务结果的队列 self.workers = []#所有的线程都存放在这个列表中 self._recruitthreads(num)#创建一系列线程的函数 def _recruitthreads(self,num): """ 创建线程 """ for i in xrange(num): worker = Worker(self.workqueue,self.resultqueue) self.workers.append(worker) def start(self): """ 启动线程池中每个线程 """ for work in self.workers: work.start() def wait_for_complete(self): """ 等待至任务队列中所有任务完成 """ while len(self.workers): worker = self.workers.pop() worker.join() if worker.isAlive() and not self.workqueue.empty(): self.workers.append(worker) def add_job(self,callable,*args,**kwargs): """ 往任务队列中添加任务 """ self.workqueue.put((callable,args,kwargs)) def get_result(self,*args,**kwargs): """ 获取结果队列 """ return self.resultqueue.get(*args,**kwargs) def add_result(self,result): self.resultqueue.put(result)
上面定义了一个线程池,它的初始化函数__init__()定义了一些存放相关数据的属性,这在Python的一些内部模块的类的定义中很常见,所有有时候多看看源码其实挺好的,学习大神的编程习惯和编程思想。
另外还要提到一点,Queue模块中的队列,不仅可以存放数据(指字符串,数值,列表,字典等等),还可以存放函数的(也就是任务),上面的代码中,callable是一个函数,当用put()将一个函数添加到队列时,put()接受的参数有函数对象以及该函数的相关参数,而且要是一个整体,所以就有了上面代码中的self.workqueue.put((callable,args,kwargs))。同理,当从这种存放函数的队列中取出数据,它返回的就是一个函数对象包括它的相关参数,有兴趣的可以打印出上面代码中run()里的callable,args,kwargs。如果你对Queue模块不了解,可参考我之前的博文,。
下面就简单的举个小例子吧。
import urllib2,datetimedef open_url(url): try: res = urllib2.urlopen(url).getcode() except urllib2.HTTPError, e: res = e.code #print res res = str(res) with open('/home/liulonghua/无标题文档','wr') as f: f.write(res) return resif __name__ == "__main__": urls = [ 'http://www.python.org', 'http://www.python.org/about/', 'http://www.onlamp.com/pub/a/python/2003/04/17/metaclasses.html', 'http://www.python.org/doc/', 'http://www.python.org/download/', 'http://www.python.org/getit/', 'http://www.python.org/community/', 'https://wiki.python.org/moin/', 'http://planet.python.org/', 'https://wiki.python.org/moin/LocalUserGroups', 'http://www.python.org/psf/', 'http://docs.python.org/devguide/', 'http://www.python.org/community/awards/' ] t1 = datetime.datetime.now() w = WorkerManger(2) for url in urls: w.add_job(open_url,url) w.start() w.wait_for_complete() t2 = datetime.datetime.now() print t2 - t1
最后结果如下:
如果把上面代码改成用多线程而不是用线程池,会是怎样的呢?
代码如下:
if __name__ == "__main__": urls = [ 'http://www.python.org', 'http://www.python.org/about/', 'http://www.onlamp.com/pub/a/python/2003/04/17/metaclasses.html', 'http://www.python.org/doc/', 'http://www.python.org/download/', 'http://www.python.org/getit/', 'http://www.python.org/community/', 'https://wiki.python.org/moin/', 'http://planet.python.org/', 'https://wiki.python.org/moin/LocalUserGroups', 'http://www.python.org/psf/', 'http://docs.python.org/devguide/', 'http://www.python.org/community/awards/' ] t1 = datetime.datetime.now() for url in urls: t = threading.Thread(target=open_url,args=(url,)) t.start() t.join() t2 = datetime.datetime.now() print t2-t1
运行结果如下:
运行效率的差异还是很大的,有兴趣的可以动手试试。
(二)使用现成的线程池模块
下载安装也很简单,用pip工具
sudo pip install threadpool
注意:这里要提到一点,我就陷入这个坑,还好没有花多长时间就解决了。由于我的电脑里有python2.7.12,python3.5,还有一个PyPy5.4.1,上面的指令竟然将threadpool包安装到了PyPy目录下了,所以在python2.7.12里,我import threadpool,它一直报错,如果你的系统里有多个Python版本,又没有用virtualenvs虚拟环境工具,很容易造成这种混乱,虽然我安装了virtualenvs,但在自己的电脑上很少用,这里的解决方法是:
sudo python -m pip install threadpool
以区分PyPy,同理如果是在PyPy环境下安装第三方包的话,用sudo pypy -m pip install packagename,这个在之前的博文中也有介绍,感兴趣的可以
该模块主要的类和方法:
1.threadpool.ThreadPool:线程池类,主要是用来分派任务请求和收集运行结果。主要方法有:
(1)__init__(self,number_workers,q_size,resq_size=0,poll_timeout=5):
建立线程池,并启动对应的num_workers的线程;q_size表示任务请求队列的大小,resq_size表示存放运行结果队列的大小。
(2)createWorkers(self,num_workers,poll_timeout=5):
将num_workers数量对应的线程加入线程池
(3)dismissWorkers(self,num_workers,do_join=False):
告诉num_workers数量的工作线程在执行完当前任务后退出
(4)joinAllDismissWorkers(self):
在设置为退出的线程上执行Thread.join
(5)putRequest(self,request,block=True,timeout=None):
加入一个任务请求到工作队列
(6)pool(self,block=False)
处理任务队列中新请求。也就是循环的调用各个线程结果中的回调和错误回调。不过,当请求队列为空时会抛出 NoResultPending 异常,以表示所有的结果都处理完了。这个特点对于依赖线程执行结果继续加入请求队列的方式不太适合。
(7)wait(self)
等待执行结果,直到所有任务完成。当所有执行结果返回后,线程池内部的线程并没有销毁,而是在等待新任务。因此,wait()之后依然可以在此调用pool.putRequest()往其中添加任务。
2. threadpool.WorkerThread:处理任务的工作线程,主要有run()方法和dismiss()方法。
3.threadpool.WorkRequest:任务请求类,包含有具体执行方法的工作请求类
__init__(self,callable,args=None,kwds=None,requestID=None,callback=None,exc_callback=None)
创建一个工作请求。
4.makeRequests(callable_,args_list,callback=None,exc_callback=_handle_thread_exception):
主要函数,用来创建具有相同的执行函数但参数不同的一系列工作请求。
有了上面自定义线程池模式的基础,这个模块不难理解,有兴趣的可以去看看该模块的源码。它的使用步骤一般如下:
(1)引入threadpool模块
(2)定义线程函数
(3)创建线程 池threadpool.ThreadPool()
(4)创建需要线程池处理的任务即threadpool.makeRequests()
(5)将创建的多个任务put到线程池中,threadpool.putRequest
(6)等到所有任务处理完毕theadpool.pool()
将上面的例子用线程池模块进行修改,代码如下:
import threadpoolif __name__ == "__main__": urls = [ 'http://www.python.org', 'http://www.python.org/about/', 'http://www.onlamp.com/pub/a/python/2003/04/17/metaclasses.html', 'http://www.python.org/doc/', 'http://www.python.org/download/', 'http://www.python.org/getit/', 'http://www.python.org/community/', 'https://wiki.python.org/moin/', 'http://planet.python.org/', 'https://wiki.python.org/moin/LocalUserGroups', 'http://www.python.org/psf/', 'http://docs.python.org/devguide/', 'http://www.python.org/community/awards/' ] t1 = datetime.datetime.now() pool = threadpool.ThreadPool(2) requests = threadpool.makeRequests(open_url,urls) [pool.putRequest(req) for req in requests] pool.wait() t2 = datetime.datetime.now() print t2-t1
执行结果如下:
该模块的其它方法,感兴趣的可以自己动手体会下。
(3)multiprocessing.dummy 执行多线程任务
multiprocessing.dummy 模块与 multiprocessing 模块的区别: dummy 模块是多线程,而 multiprocessing 是多进程, api 都是通用的。
Python3里的multiprocessing里也有现成的线程池,如下
from multiprocessing.pool import ThreadPool
有时候看到有人这么用dummy,from multiprocessing.dummy import Pool as ThreadPool ,把它当作了一个线程池。它的属性和方法可以参考。将上面的例子可以用这种方法改下代码如下:
from multiprocessing.dummy import Pool as ThreadPool if __name__ == "__main__": urls = [ 'http://www.python.org', 'http://www.python.org/about/', 'http://www.onlamp.com/pub/a/python/2003/04/17/metaclasses.html', 'http://www.python.org/doc/', 'http://www.python.org/download/', 'http://www.python.org/getit/', 'http://www.python.org/community/', 'https://wiki.python.org/moin/', 'http://planet.python.org/', 'https://wiki.python.org/moin/LocalUserGroups', 'http://www.python.org/psf/', 'http://docs.python.org/devguide/', 'http://www.python.org/community/awards/' ] t1 = datetime.datetime.now() pool =ThreadPool(2) pool.map(open_url,urls) pool.close() pool.join() t2 = datetime.datetime.now() print t2-t1
运行结果如下:
我觉得上面三种方法的主体思路还是差不多的,还是比较好理解的,希望对你有帮助,不正之处欢迎批评指正!