python线程池实现
Posted rianley 的博客
tags:
篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了python线程池实现相关的知识,希望对你有一定的参考价值。
简单实现一个线程池:
import queue import threading import time class ThreadPool(object): #创建线程池类 def __init__(self, max_num=20): #创建一个最大长度为20的队列 self.queue = queue.Queue(max_num) #创建一个队列 for i in range(max_num): #循环把线程对象加入到队列中 self.queue.put(threading.Thread) #把线程的类名放进去,执行完这个Queue def get_thread(self): #定义方法从队列里获取线程 return self.queue.get() #在队列中获取值 def add_thread(self): #线程执行完任务后,在队列里添加线程 self.queue.put(threading.Thread) def func(pool,a1): time.sleep(1) print(a1) pool.add_thread() #线程执行完任务后,队列里再加一个线程 p = ThreadPool(10) #执行init方法; 一次最多执行10个线程 for i in range(100): thread = p.get_thread() #线程池10个线程,每一次循环拿走一个拿到类名,没有就等待 t = thread(target=func, args=(p, i,)) #创建线程; 线程执行func函数的这个任务;args是给函数传入参数 t.start() #激活线程
复杂线程池
线程池要点:
1,创建线程池时,是在需要执行线程的时候创建线程,而不是创建好最大队列等待执行
2,创建一个回调函数,检查出剩余队列的任务,当线程执行完函数的时候通知线程池,
3,使用线程池时让其循环获取任务,并执行
4,线程池,让其自行的去激活线程,执行完成后,关闭退出
import queue import threading import time import contextlib StopEvent = object() class ThreadPool(object): def __init__(self, max_num): self.q = queue.Queue() # 最多创建的线程数(线程池最大容量) self.max_num = max_num self.terminal = False #如果为True 终止所有线程,不在获取新任务 self.generate_list = [] # 真实创建的线程列表 self.free_list = []# 空闲线程数量 def run(self, func, args, callback=None): """ 线程池执行一个任务 :param func: 任务函数 :param args: 任务函数所需参数 :param callback: 任务执行失败或成功后执行的回调函数,回调函数有两个参数1、任务函数执行状态;2、任务函数返回值(默认为None,即:不执行回调函数) :return: 如果线程池已经终止,则返回True否则None """ if len(self.free_list) == 0 and len(self.generate_list) < self.max_num: self.generate_thread() #创建线程 w = (func, args, callback,) #把参数封装成元祖 self.q.put(w) #添加到任务队列 def generate_thread(self): """ 创建一个线程 """ t = threading.Thread(target=self.call) t.start() def call(self): """ 循环去获取任务函数并执行任务函数 """ current_thread = threading.currentThread # 获取当前线程 self.generate_list.append(current_thread) #添加到已经创建的线程里 event = self.q.get() # 取任务并执行 while event != StopEvent: # 是元组=》是任务;如果不为停止信号 执行任务 func, arguments, callback = event #解开任务包; 分别取出值 try: result = func(*arguments) #运行函数,把结果赋值给result status = True #运行结果是否正常 except Exception as e: status = False #表示运行不正常 result = e #结果为错误信息 if callback is not None: #是否存在回调函数 try: callback(status, result) #执行回调函数 except Exception as e: pass if self.terminal: # 默认为False,如果调用terminal方法 event = StopEvent #等于全局变量,表示停止信号 else: # self.free_list.append(current_thread) #执行完毕任务,添加到闲置列表 # event = self.q.get() #获取任务 # self.free_list.remove(current_thread) # 获取到任务之后,从闲置列表中删除;不是元组,就不是任务 with self.worker_state(self.free_list, current_thread): event = self.q.get() else: self.generate_list.remove(current_thread) #如果收到终止信号,就从已经创建的线程列表中删除 def close(self): #终止线程 num = len(self.generate_list) #获取总共创建的线程数 while num: self.q.put(StopEvent) #添加停止信号,有多少线程添加多少表示终止的信号 num -= 1 def terminate(self): #终止线程(清空队列) self.terminal = True #把默认的False更改成True while self.generate_list: #如果有已经创建线程存活 self.q.put(StopEvent) #有几个线程就发几个终止信号 self.q.empty() #清空队列 @contextlib.contextmanager def worker_state(self, state_list, worker_thread): state_list.append(worker_thread) try: yield finally: state_list.remove(worker_thread) def work(i): print(i) pool = ThreadPool(10) for item in range(50): pool.run(func=work, args=(item,)) # 将任务放在队列中 # 着手开始处理任务 # - 创建线程 # - 有空闲线程,择不再创建线程 # - 不能高于线程池的限制 # - 根据任务个数判断 # - 线程去队列中取任务 pool.terminate()
本文参考oldboy吴佩琦老师(详细地址:)http://www.cnblogs.com/wupeiqi/articles/4839959.html
以上是关于python线程池实现的主要内容,如果未能解决你的问题,请参考以下文章
Motan在服务provider端用于处理request的线程池