python多线程抽象编程模型详解
最近需要完成一个多线程下载的工具,对其中的多线程下载进行了一个抽象,可以对所有需要使用到多线程编程的地方统一使用这个模型来进行编写。
主要结构:
1、基于Queue标准库实现了一个类似线程池的工具,用户指定提交任务线程submitter与工作线程worker数目,所有线程分别设置为后台运行,提供等待线程运行完成的接口。
2、所有需要完成的任务抽象成task,提供单独的无参数调用方式,供worker线程调用;task以生成器的方式作为参数提供,供submitter调用。
3、所有需要进行线程交互的信息放在context类中。
主要实现代码如下:
#Submitter线程类实现,主要是`task_generator`调用 classSubmitterThread(threading.Thread): _DEFAULT_WAIT_TIMEOUT=2#seconds def__init__(self,queue,task_gen,timeout=2): super(SubmitterThread,self).__init__() self.queue=queue ifnotisinstance(timeout,int): _logger.error('Threadwaittimeoutvalueerror:%s,' 'usedefaultinstead.'%timeout) self.timeout=self._DEFAULT_WAIT_TIMEOUT self.timeout=timeout self.task_generator=task_gen defrun(self): whileTrue: try: task=self.task_generator.next() self.queue.put(task,True,self.timeout) exceptQueue.Full: _logger.debug('Taskqueueisfull.%swait%dsecond%stimeout'% (self.name,self.timeout,'s'if(self.timeout>1)else'')) break except(StopIteration,ValueError)ase: _logger.debug('Taskfinished') break
#Worker线程实现,主要就是try块内的func调用 classWorkerThread(threading.Thread): _DEFAULT_WAIT_TIMEOUT=2#seconds def__init__(self,queue,timeout=2): super(WorkerThread,self).__init__() self.queue=queue ifnotisinstance(timeout,int): _logger.error('Threadwaittimeoutvalueerror:%s,' 'usedefaultinstead.'%timeout) self.timeout=self._DEFAULT_WAIT_TIMEOUT self.timeout=timeout defrun(self): whileTrue: try: func=self.queue.get(True,self.timeout) exceptQueue.Empty: _logger.debug('Taskqueueisempty.%swait%dsecond%stimeout'% (self.name,self.timeout,'s'if(self.timeout>1)else'')) break ifnotcallable(func): time.sleep(1) try: func() exceptExceptionase: _logger.error('Thread%srunningoccurserror:%s'% (self.name,e)) print('Threadrunningerror:%s'%e)
classExecutor(object): """ Thereallyplacetoexecuteexecutor """ thread_list=[] submitters=0 workers=0 queue=None task_generator=None timeout=0 def__init__(self,task_gen,submitters=1,workers=1,timeout=2): iflen(self.thread_list)!=0: raiseRuntimeError('Executorcanonlyinstanceonce.') self.queue=Queue.Queue(maxsize=submitters*2+workers*2) self.submitters=submitters self.workers=workers self.task_generator=task_gen self.timeout=timeout defstart(self): foriinrange(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() foriinrange(self.workers): worker=WorkerThread(self.queue,self.timeout) self.thread_list.append(worker) worker.setName('Worker-%d'%i) worker.setDaemon(True) worker.start() defis_alive(self): alive=False fortinself.thread_list: ift.isAlive(): alive=True break returnalive defwait_to_shutdown(self): _logger.debug('Starttowaittoshutdown') fortinself.thread_list: t.join() _logger.debug('Shutdownthread:%s'%t.name)
Executor类保存了线程池,提供相应接口。有了这个抽象之后,只需要实例化Executor类的对象,然后调用start方法进行多线程任务的运行。并可以用is_alive等接口再主线程内进行其他处理。
后续再使用这个抽象进行实际多线程任务的实现。
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持毛票票。