#coding:gbk
import Queue
import threading
import sys
class WorkerManager:
def __init__(self, num_of_workers=10, timeout = 1):
self.workQueue = Queue.Queue()
self.resultQueue = Queue.Queue()
self.threads = []
self.timeout = timeout
self._recruitThreads(num_of_workers)
def _recruitThreads(self, num_of_workers):
for i in range(num_of_workers):
thread = Worker(self.workQueue, self.resultQueue, self.timeout)
self.threads.append(thread)
def wait_for_complete(self):
while len(self.threads):
thread = self.threads.pop()
thread.join( )
if thread.isAlive() and not self.workQueue.empty():
self.threads.append(thread)
def add_job(self, callable, *args, **kwds ):
self.workQueue.put((callable, args, kwds))
def get_result(self, *args, **kwds):
return self.resultQueue.get(*args, **kwds)
class Worker(threading.Thread):
def __init__( self, workQueue, resultQueue, timeout = 0, **kwds):
threading.Thread.__init__( self, **kwds )
self.setDaemon(True)
self.workQueue = workQueue
self.resultQueue = resultQueue
self.timeout = timeout
self.start()
def run( self ):
while True:
try:
callable, args, kwds = self.workQueue.get(timeout=self.timeout)
res = callable(*args, **kwds)
self.resultQueue.put(res)
except Queue.Empty:
break
except:
print 'Worker: %s %s %s' % sys.exc_info()
def func(index, fname):
print index, fname
if __name__=='__main__':
flist = [1,2,3,4,5,6,7,8,9,10]
worker_num = 2
wm = WorkerManager(worker_num)
for i in range(len(flist)):
fname = flist[i]
wm.add_job(func, i, fname)
wm.wait_for_complete()