线程池:
方案简介:
方案一:简单版本的线程池,每次都要创建线程池;
方案二:支持传函数、传参、传回调函数、立即终止所有线程、最大优点:线程的循环利用,节省时间和资源 ★★★★★
方案三:现有模块,直接调用即可,不支持回调函数
方案一:
#!/usr/bin/env python # -*- coding:utf-8 -*- import Queue import threading class ThreadPool(object): def __init__(self, max_num=20): self.queue = Queue.Queue(max_num) for i in xrange(max_num): self.queue.put(threading.Thread) def get_thread(self): return self.queue.get() def add_thread(self): self.queue.put(threading.Thread) """ pool = ThreadPool(10) def func(arg, p): print arg import time time.sleep(2) p.add_thread() for i in xrange(30): thread = pool.get_thread() t = thread(target=func, args=(i, pool)) t.start() """
方案二:
#!/usr/bin/env python # -*- coding:utf-8 -*- import queue import threading import contextlib import time StopEvent = object() class ThreadPool(object): def __init__(self, max_num): self.q = queue.Queue() self.max_num = max_num self.terminal = False 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) 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 event = StopEvent else: with self.worker_state(self.free_list,current_thread): event = self.q.get() else: self.generate_list.remove(current_thread) @contextlib.contextmanager def worker_state(self,x,v): x.append(v) try: yield finally: x.remove(v) def close(self): num = len(self.generate_list) while num: self.q.put(StopEvent) num -= 1 # 终止线程(清空队列) def terminate(self): self.terminal = True while self.generate_list: self.q.put(StopEvent) self.q.empty() import time def work(i): time.sleep(1) print(i) pool = ThreadPool(10) for item in range(50): pool.run(func=work, args=(item,)) # pool.terminate() #立即终止所有线程
方案三、
from concurrent.futures import ThreadPoolExecutor import time def f1(a): time.sleep(2) print(a) return 1 pool=ThreadPoolExecutor(5) for i in range(30): a=pool.submit(f1,i) # x=a.result()#获取返回值,如果有,会阻塞
时间: 2024-10-03 13:40:00