import sys
import threading
import Queue
import traceback
# 定义一些Exception，用于自定义异常处理
class NoResultsPending(Exception):
"""All works requests have been processed"""
pass
class NoWorkersAvailable(Exception):
"""No worket threads available to process remaining requests."""
pass
def _handle_thread_exception(request, exc_info):
"""默认的异常处理函数，只是简单的打印"""
traceback.print_exception(*exc_info)
#classes
class WorkerThread(threading.Thread):
"""后台线程，真正的工作线程，从请求队列(requestQueue)中获取work，
并将执行后的结果添加到结果队列(resultQueue)"""
def __init__(self,requestQueue,resultQueue,poll_timeout=5,**kwds):
threading.Thread.__init__(self,**kwds)
'''设置为守护进行'''
self.setDaemon(True)
self._requestQueue = requestQueue
self._resultQueue = resultQueue
self._poll_timeout = poll_timeout
'''设置一个flag信号，用来表示该线程是否还被dismiss,默认为false'''
self._dismissed = threading.Event()
self.start()
def run(self):
'''每个线程尽可能多的执行work，所以采用loop，
只要线程可用，并且requestQueue有work未完成，则一直loop'''
while True:
if self._dismissed.is_set():
break
try:
'''
Queue.Queue队列设置了线程同步策略，并且可以设置timeout。
一直block，直到requestQueue有值，或者超时
'''
request = self._requestQueue.get(True,self._poll_timeout)
except Queue.Empty:
continue
else:
'''之所以在这里再次判断dimissed，是因为之前的timeout时间里，很有可能，该线程被dismiss掉了'''
if self._dismissed.is_set():
self._requestQueue.put(request)
break
try:
'''执行callable，讲请求和结果以tuple的方式放入requestQueue'''
result = request.callable(*request.args,**request.kwds)
print self.getName()
self._resultQueue.put((request,result))
except:
'''异常处理'''
request.exception = True
self._resultQueue.put((request,sys.exc_info()))
def dismiss(self):
'''设置一个标志，表示完成当前work之后，退出'''
self._dismissed.set()
class WorkRequest:
'''
@param callable_:，可定制的，执行work的函数
@param args: 列表参数
@param kwds: 字典参数
@param requestID: id
@param callback: 可定制的，处理resultQueue队列元素的函数
@param exc_callback:可定制的，处理异常的函数
'''
def __init__(self,callable_,args=None,kwds=None,requestID=None,
callback=None,exc_callback=_handle_thread_exception):
if requestID == None:
self.requestID = id(self)
else:
try:
self.requestID = hash(requestID)
except TypeError:
raise TypeError("requestId must be hashable")
self.exception = False
self.callback = callback
self.exc_callback = exc_callback
self.callable = callable_
self.args = args or []
self.kwds = kwds or {}
def __str__(self):
return "WorkRequest id=%s args=%r kwargs=%r exception=%s" %
(self.requestID,self.args,self.kwds,self.exception)
class ThreadPool:
'''
@param num_workers:初始化的线程数量
@param q_size,resq_size: requestQueue和result队列的初始大小
@param poll_timeout: 设置工作线程WorkerThread的timeout，也就是等待requestQueue的timeout
'''
def __init__(self,num_workers,q_size=0,resq_size=0,poll_timeout=5):
self._requestQueue = Queue.Queue(q_size)
self._resultQueue = Queue.Queue(resq_size)
self.workers = []
self.dismissedWorkers = []
self.workRequests = {} #设置个字典，方便使用
self.createWorkers(num_workers,poll_timeout)
def createWorkers(self,num_workers,poll_timeout=5):
'''创建num_workers个WorkThread,默认timeout为5'''
for i in range(num_workers):
self.workers.append(WorkerThread(self._requestQueue,self._resultQueue,poll_timeout=poll_timeout))
def dismissWorkers(self,num_workers,do_join=False):
'''停用num_workers数量的线程，并加入dismiss_list'''
dismiss_list = []
for i in range(min(num_workers,len(self.workers))):
worker = self.workers.pop()
worker.dismiss()
dismiss_list.append(worker)
if do_join :
for worker in dismiss_list:
worker.join()
else:
self.dismissedWorkers.extend(dismiss_list)
def joinAllDismissedWorkers(self):
'''join 所有停用的thread'''
#print len(self.dismissedWorkers)
for worker in self.dismissedWorkers:
worker.join()
self.dismissedWorkers = []
def putRequest(self,request ,block=True,timeout=None):
assert isinstance(request,WorkRequest)
assert not getattr(request,'exception',None)
'''当queue满了，也就是容量达到了前面设定的q_size,它将一直阻塞，直到有空余位置，或是timeout'''
self._requestQueue.put(request, block, timeout)
self.workRequests[request.requestID] = request
def poll(self,block = False):
while True:
if not self.workRequests:
raise NoResultsPending
elif block and not self.workers:
raise NoWorkersAvailable
try:
'''默认只要resultQueue有值，则取出，否则一直block'''
request , result = self._resultQueue.get(block=block)
if request.exception and request.exc_callback:
request.exc_callback(request,result)
if request.callback and not (request.exception and request.exc_callback):
request.callback(request,result)
del self.workRequests[request.requestID]
except Queue.Empty:
break
def wait(self):
while True:
try:
self.poll(True)
except NoResultsPending:
break
def workersize(self):
return len(self.workers)
def stop(self):
'''join 所有的thread,确保所有的线程都执行完毕'''
self.dismissWorkers(self.workersize(),True)
self.joinAllDismissedWorkers()
if __name__=='__main__':
import random
import time
import datetime
def do_work(data):
time.sleep(random.randint(1,3))
res = str(datetime.datetime.now()) + "" +str(data)
return res
def print_result(request,result):
print "---Result from request %s : %r" % (request.requestID,result)
main = ThreadPool(3)
for i in range(40):
req = WorkRequest(do_work,args=[i],kwds={},callback=print_result)
main.putRequest(req)
print "work request #%s added." % req.requestID
print '-'*20, main.workersize(),'-'*20
counter = 0
while True:
try:
time.sleep(0.5)
main.poll()
if(counter==5):
print "Add 3 more workers threads"
main.createWorkers(3)
print '-'*20, main.workersize(),'-'*20
if(counter==10):
print "dismiss 2 workers threads"
main.dismissWorkers(2)
print '-'*20, main.workersize(),'-'*20
counter+=1
except NoResultsPending:
print "no pending results"
break
main.stop()
print "Stop"
