快捷導(dǎo)航

python多線程抽象編程模型詳解

更新時(shí)間：2019年03月20日 10:14:25 作者：OshynSong

這篇文章主要為大家詳細(xì)介紹了python多線程抽象編程模型，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下

最近需要完成一個多線程下載的工具，對其中的多線程下載進(jìn)行了一個抽象，可以對所有需要使用到多線程編程的地方統(tǒng)一使用這個模型來進(jìn)行編寫。

主要結(jié)構(gòu)：

1、基于Queue標(biāo)準(zhǔn)庫實(shí)現(xiàn)了一個類似線程池的工具，用戶指定提交任務(wù)線程submitter與工作線程worker數(shù)目，所有線程分別設(shè)置為后臺運(yùn)行，提供等待線程運(yùn)行完成的接口。

2、所有需要完成的任務(wù)抽象成task，提供單獨(dú)的無參數(shù)調(diào)用方式，供worker線程調(diào)用；task以生成器的方式作為參數(shù)提供，供submitter調(diào)用。

3、所有需要進(jìn)行線程交互的信息放在context類中。

主要實(shí)現(xiàn)代碼如下:

#Submitter線程類實(shí)現(xiàn)，主要是`task_generator`調(diào)用
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線程實(shí)現(xiàn)，主要就是try塊內(nèi)的func調(diào)用
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)

Executor類保存了線程池，提供相應(yīng)接口。有了這個抽象之后，只需要實(shí)例化Executor類的對象，然后調(diào)用start方法進(jìn)行多線程任務(wù)的運(yùn)行。并可以用is_alive等接口再主線程內(nèi)進(jìn)行其他處理。

后續(xù)再使用這個抽象進(jìn)行實(shí)際多線程任務(wù)的實(shí)現(xiàn)。

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

您可能感興趣的文章: