python多线程抽象编程模型详解

#Submitter线程类实现，主要是`task_generator`调用class SubmitterThread(threading.Thread):  _DEFAULT_WAIT_TIMEOUT = 2 #seconds  def __init__(self, queue, task_gen, timeout=2):    super(SubmitterThread, self).__init__()    self.queue = queue    if not isinstance(timeout, int):      _logger.error('Thread wait timeout value error: %s, '             'use default instead.' % timeout)      self.timeout = self._DEFAULT_WAIT_TIMEOUT    self.timeout = timeout    self.task_generator = task_gen  def run(self):    while True:      try:        task = self.task_generator.next()        self.queue.put(task, True, self.timeout)      except Queue.Full:        _logger.debug('Task queue is full. %s wait %d second%s timeout' %               (self.name, self.timeout, 's' if (self.timeout > 1) else ''))        break      except (StopIteration, ValueError) as e:        _logger.debug('Task finished')        break

#Worker线程实现，主要就是try块内的func调用class WorkerThread(threading.Thread):  _DEFAULT_WAIT_TIMEOUT = 2 #seconds  def __init__(self, queue, timeout=2):    super(WorkerThread, self).__init__()    self.queue = queue    if not isinstance(timeout, int):      _logger.error('Thread wait timeout value error: %s, '             'use default instead.' % timeout)      self.timeout = self._DEFAULT_WAIT_TIMEOUT    self.timeout = timeout  def run(self):    while True:      try:        func = self.queue.get(True, self.timeout)      except Queue.Empty:        _logger.debug('Task queue is empty. %s wait %d second%s timeout' %               (self.name, self.timeout, 's' if (self.timeout > 1) else ''))        break      if not callable(func):        time.sleep(1)      try:        func()      except Exception as e:        _logger.error('Thread %s running occurs error: %s' %               (self.name, e))        print('Thread running error: %s' % e)

class Executor(object):  """  The really place to execute executor  """  thread_list = []  submitters = 0  workers = 0  queue = None  task_generator = None  timeout = 0  def __init__(self, task_gen, submitters=1, workers=1 , timeout=2):    if len(self.thread_list) != 0:      raise RuntimeError('Executor can only instance once.')    self.queue = Queue.Queue(maxsize=submitters * 2 + workers * 2)    self.submitters = submitters    self.workers = workers    self.task_generator = task_gen    self.timeout = timeout  def start(self):    for i in range(self.submitters):      submitter = SubmitterThread(self.queue, self.task_generator, self.timeout)      self.thread_list.append(submitter)      submitter.setName('Submitter-%d' % i)      submitter.setDaemon(True)      submitter.start()    for i in range(self.workers):      worker = WorkerThread(self.queue, self.timeout)      self.thread_list.append(worker)      worker.setName('Worker-%d' % i)      worker.setDaemon(True)      worker.start()  def is_alive(self):    alive = False    for t in self.thread_list:      if t.isAlive():        alive = True        break    return alive  def wait_to_shutdown(self):    _logger.debug('Start to wait to shutdown')    for t in self.thread_list:      t.join()      _logger.debug('Shutdown thread : %s' % t.name)