最近需要完成一个多线程下载的工具，对其中的多线程下载进行了一个抽象，可以对所有需要使用到多线程编程的地方统一使用这个模型来进行编写。
主要结构：
1、基于Queue标准库实现了一个类似线程池的工具，用户指定提交任务线程submitter与工作线程worker数目，所有线程分别设置为后台运行，提供等待线程运行完成的接口。
2、所有需要完成的任务抽象成task，提供单独的无参数调用方式，供worker线程调用；task以生成器的方式作为参数提供，供submitter调用。
3、所有需要进行线程交互的信息放在context类中。
主要实现代码如下:
#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)
Executor类保存了线程池，提供相应接口。有了这个抽象之后，只需要实例化Executor类的对象，然后调用start方法进行多线程任务的运行。并可以用is_alive等接口再主线程内进行其他处理。
后续再使用这个抽象进行实际多线程任务的实现。
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。
免责声明：本站文章均来自网站采集或用户投稿，网站不提供任何软件下载或自行开发的软件！ 如有用户或公司发现本站内容信息存在侵权行为，请邮件告知！ 858582#qq.com
《魔兽世界》大逃杀！60人新游玩模式《强袭风暴》3月21日上线
暴雪近日发布了《魔兽世界》10.2.6 更新内容，新游玩模式《强袭风暴》即将于3月21 日在亚服上线，届时玩家将前往阿拉希高地展开一场 60 人大逃杀对战。
艾泽拉斯的冒险者已经征服了艾泽拉斯的大地及遥远的彼岸。他们在对抗世界上最致命的敌人时展现出过人的手腕，并且成功阻止终结宇宙等级的威胁。当他们在为即将于《魔兽世界》资料片《地心之战》中来袭的萨拉塔斯势力做战斗准备时，他们还需要在熟悉的阿拉希高地面对一个全新的敌人──那就是彼此。在《巨龙崛起》10.2.6 更新的《强袭风暴》中，玩家将会进入一个全新的海盗主题大逃杀式限时活动，其中包含极高的风险和史诗级的奖励。
《强袭风暴》不是普通的战场，作为一个独立于主游戏之外的活动，玩家可以用大逃杀的风格来体验《魔兽世界》，不分职业、不分装备（除了你在赛局中捡到的），光是技巧和战略的强弱之分就能决定出谁才是能坚持到最后的赢家。本次活动将会开放单人和双人模式，玩家在加入海盗主题的预赛大厅区域前，可以从强袭风暴角色画面新增好友。游玩游戏将可以累计名望轨迹，《巨龙崛起》和《魔兽世界：巫妖王之怒 经典版》的玩家都可以获得奖励。