自定义进程池的方法

Posted 2020-09-18 叶祖辉

一 、比较low的线程池

import queue,time,threading
class My_theading_pool(object):
    def __init__(self,num = 20):
        self.queue = queue.Queue(num) #在类中分装一个队列，队列中最多容纳20
        for i in range(num):
            self.queue.put(threading.Thread) #在队列的20个位置上放置线程
            
    def get_thead(self):
        return self.queue.get()   #从队列中去取值
    
    def add_thead(self):  #线程用完以后，再往队列中添加线程
        return self.queue.put(threading.Thread)

def func(pool,a1):
    time.sleep(1)
    print(a1)
    pool.add_thead()            

t = My_theading_pool(10)  #创建一个里面有10个线程的线程池
for i in range(100):      #创建100个任务
    t1 = t.get_thead()    #每一次任务执行的时候都会去队列中去取出来一个线程
    t2 = t1(target = func,args = (t,i))  #取出来的每一个线程都执行一个func函数，并把类对象当做参数传入函数中，函数中再执行add方法　　                
    t2.start()


注：定义一个类，创建一个类对象，在对象中封装一个队列，定义队列的容量，并利用循环往队列中添加线程，利用循环创建任务，并从队列中获取线程执行函数，并把类对象传入函数中
   执行完一个任务后，在利用传进来的类对象执行add函数获取一个线程执行任务，执行完毕后再往队列中添加新的线程

 二 、比较高大上的线程池

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

            self.free_list.append(current_thread)
            event = self.q.get()
            self.free_list.remove(current_thread)
        else:
            self.generate_list.remove(current_thread)

    def close(self):
        num = len(self.generate_list)
        while num:
            self.q.put(StopEvent)
            num -= 1


import time

def work(i):
    print(i)

pool = ThreadPool(10)
for item in range(50):
    pool.run(func=work, args=(item,))

pool.close()