threadpool源码学习
__all__ = [
'makeRequests',
'NoResultsPending',
'NoWorkersAvailable',
'ThreadPool',
'WorkRequest',
'WorkerThread'
]
WorkerThread
WorkerThread(requests_queue, results_queue) 将实例化一个线程对象，实例化后就会执行自身的run方法。run方法里执行一个while循环，一直去requests_queue取任务(WorkRequest对象)，如果获取到就执行WorkRequest对象的callable(*request.args, **request.kwds)方法并将返回赋值给result, 然后把结果放到_results_queue里，放进去格式为元组(request, result)，如果执行过程中遇到异常就将request.exception设置为True并将异常信息放入。dismiss方法通过将Event对象设置为True的方式将run中的while循环break掉。代码如下：
class WorkerThread(threading.Thread):
def __init__(self, requests_queue, results_queue, poll_timeout=5, **kwds):
threading.Thread.__init__(self, **kwds)
self.setDaemon(1)
self._requests_queue = requests_queue
self._results_queue = results_queue
self._poll_timeout = poll_timeout
self._dismissed = threading.Event()
self.start()
def run(self):
while True:
if self._dismissed.isSet():
break
try:
request = self._requests_queue.get(True)
except Queue.Empty:
continue
else:
if self._dismissed.isSet():
self._requests_queue.put(request)
break
try:
result = request.callable(*request.args, **request.kwds)
self._results_queue.put((request, result))
except:
request.exception = True
self._results_queue.put((request, sys.exc_info()))
def dismiss(self):
self._dismissed.set()
WorkRequest
WorkRequest()将实例化一个request对象，实例化时会设置一个可hash的requestID值。callable_是将要被线程执行的可callable的对象，callback是执行完任务后执行的回调方法，exc_callback是任务执行异常时的回调。代码如下：
class WorkRequest:
def __init__(self, callable_, args=None, kwds=None, requestID=None,
callback=None, exc_callback=_handle_thread_exception):
if requestID is 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)
makeRequests
makeRequests是一个函数，makeRequests()调用后会返回一个requests列表，这里args_list里元素有两种，如果只有位置参数那么传入request对象时就是(位置参数)，如果既有位置参数又有关键字参数那么传入request对象就是(位置参数)+ {关键字参数字典}。代码如下：
def makeRequests(callable_, args_list, callback=None,
exc_callback=_handle_thread_exception):
requests = []
for item in args_list:
if isinstance(item, tuple):
requests.append(
WorkRequest(callable_, item[0], item[1], callback=callback,
exc_callback=exc_callback)
)
else:
requests.append(
WorkRequest(callable_, [item], None, callback=callback,
exc_callback=exc_callback)
)
return requests
ThreadPool
ThreadPool(num_workers)将实例化一个有num_workers数量线程的线程池对象，线程池的线程共用_requests_queue和_results_queue两个队列，当执行putRequest(request)时就会将request放入_requests_queue这个队列，这时候池内的线程就能从_requests_queue获取到值并执行了，要获取结果需要执行poll()方法取到一个结果就会将workRequests{requestID: request}这个字典里request删除掉，最后取完后就会有NoResultsPending异常。这里如果调用wait方法将会等待所有结果。代码如下：
class ThreadPool:
def __init__(self, num_workers, q_size=0, resq_size=0, poll_timeout=5):
self._requests_queue = Queue.Queue(q_size)
self._results_queue = Queue.Queue(resq_size)
self.workers = []
self.dismissedWorkers = []
self.workRequests = {}
self.createWorkers(num_workers, poll_timeout)
def createWorkers(self, num_workers, poll_timeout=5):
for i in range(num_workers):
self.workers.append(WorkerThread(self._requests_queue,
self._results_queue, poll_timeout=poll_timeout))
def dismissWorkers(self, num_workers, do_join=False):
"""Tell num_workers worker threads to quit after their current task."""
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):
"""Perform Thread.join() on all worker threads that have been dismissed.
"""
for worker in self.dismissedWorkers:
worker.join()
self.dismissedWorkers = []
def putRequest(self, request, block=True, timeout=None):
"""Put work request into work queue and save its id for later."""
assert isinstance(request, WorkRequest)
# don't reuse old work requests
assert not getattr(request, 'exception', None)
self._requests_queue.put(request, block, timeout)
self.workRequests[request.requestID] = request
def poll(self, block=False):
"""Process any new results in the queue."""
while True:
# still results pending?
if not self.workRequests:
raise NoResultsPending
# are there still workers to process remaining requests?
elif block and not self.workers:
raise NoWorkersAvailable
try:
# get back next results
request, result = self._results_queue.get(block=block)
# has an exception occured?
if request.exception and request.exc_callback:
request.exc_callback(request, result)
# hand results to callback, if any
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):
"""Wait for results, blocking until all have arrived."""
while 1:
try:
self.poll(True)
except NoResultsPending:
break
源码中的测试example
我觉得这个example太舒服了，看注释就差不多都能看懂了，就不多说了，毕竟talk is cheap！
if __name__ == '__main__':
import random
import time
# the work the threads will have to do (rather trivial in our example)
def do_something(data):
time.sleep(random.randint(1,5))
result = round(random.random() * data, 5)
# just to show off, we throw an exception once in a while
if result > 5:
raise RuntimeError("Something extraordinary happened!")
return result
# this will be called each time a result is available
def print_result(request, result):
print("**** Result from request #%s: %r" % (request.requestID, result))
# this will be called when an exception occurs within a thread
# this example exception handler does little more than the default handler
def handle_exception(request, exc_info):
if not isinstance(exc_info, tuple):
# Something is seriously wrong...
print(request)
print(exc_info)
raise SystemExit
print("**** Exception occured in request #%s: %s" % \
(request.requestID, exc_info))
# assemble the arguments for each job to a list...
data = [random.randint(1,10) for i in range(20)]
# ... and build a WorkRequest object for each item in data
requests = makeRequests(do_something, data, print_result, handle_exception)
# to use the default exception handler, uncomment next line and comment out
# the preceding one.
#requests = makeRequests(do_something, data, print_result)
# or the other form of args_lists accepted by makeRequests: ((,), {})
data = [((random.randint(1,10),), {}) for i in range(20)]
requests.extend(
makeRequests(do_something, data, print_result, handle_exception)
#makeRequests(do_something, data, print_result)
# to use the default exception handler, uncomment next line and comment
# out the preceding one.
)
# we create a pool of 3 worker threads
print("Creating thread pool with 3 worker threads.")
main = ThreadPool(3)
# then we put the work requests in the queue...
for req in requests:
main.putRequest(req)
print("Work request #%s added." % req.requestID)
# or shorter:
# [main.putRequest(req) for req in requests]
# ...and wait for the results to arrive in the result queue
# by using ThreadPool.wait(). This would block until results for
# all work requests have arrived:
# main.wait()
# instead we can poll for results while doing something else:
i = 0
while True:
try:
time.sleep(0.5)
main.poll()
print("Main thread working...")
print("(active worker threads: %i)" % (threading.activeCount()-1, ))
if i == 10:
print("**** Adding 3 more worker threads...")
main.createWorkers(3)
if i == 20:
print("**** Dismissing 2 worker threads...")
main.dismissWorkers(2)
i += 1
except KeyboardInterrupt:
print("**** Interrupted!")
break
except NoResultsPending:
print("**** No pending results.")
break
if main.dismissedWorkers:
print("Joining all dismissed worker threads...")
main.joinAllDismissedWorkers()