您好,登錄后才能下訂單哦!
這篇文章主要為大家展示了“python多線程抽象編程模型的示例分析”,內(nèi)容簡而易懂,條理清晰,希望能夠幫助大家解決疑惑,下面讓小編帶領(lǐng)大家一起研究并學習一下“python多線程抽象編程模型的示例分析”這篇文章吧。
主要結(jié)構(gòu):
1、基于Queue標準庫實現(xiàn)了一個類似線程池的工具,用戶指定提交任務(wù)線程submitter與工作線程worker數(shù)目,所有線程分別設(shè)置為后臺運行,提供等待線程運行完成的接口。
2、所有需要完成的任務(wù)抽象成task,提供單獨的無參數(shù)調(diào)用方式,供worker線程調(diào)用;task以生成器的方式作為參數(shù)提供,供submitter調(diào)用。
3、所有需要進行線程交互的信息放在context類中。
主要實現(xiàn)代碼如下:
#Submitter線程類實現(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線程實現(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類保存了線程池,提供相應接口。有了這個抽象之后,只需要實例化Executor類的對象,然后調(diào)用start方法進行多線程任務(wù)的運行。并可以用is_alive等接口再主線程內(nèi)進行其他處理。
后續(xù)再使用這個抽象進行實際多線程任務(wù)的實現(xiàn)。
以上是“python多線程抽象編程模型的示例分析”這篇文章的所有內(nèi)容,感謝各位的閱讀!相信大家都有了一定的了解,希望分享的內(nèi)容對大家有所幫助,如果還想學習更多知識,歡迎關(guān)注億速云行業(yè)資訊頻道!
免責聲明:本站發(fā)布的內(nèi)容(圖片、視頻和文字)以原創(chuàng)、轉(zhuǎn)載和分享為主,文章觀點不代表本網(wǎng)站立場,如果涉及侵權(quán)請聯(lián)系站長郵箱:is@yisu.com進行舉報,并提供相關(guān)證據(jù),一經(jīng)查實,將立刻刪除涉嫌侵權(quán)內(nèi)容。