python基础-第九篇-9.3线程池

简单版

　　

import queue
import threading


class ThreadPool(object):

    def __init__(self, max_num=20):
        self.queue = queue.Queue(max_num)
        for i in range(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)
#执行构造方法--实例一个max_num=10的队列，并往队列上添加10个线程类


def func(arg, p):
    print(arg)
    import time
    time.sleep(2)
    #在线程执行完前往队列里再加上一个线程类
    p.add_thread()

#生成30个任务
for i in range(30):
    #每生成一个任务就从队列里取出一个线程类
    thread = pool.get_thread()
    #并实例化线程对象来执行任务
    t = thread(target=func, args=(i, pool))
    #线程启动后，执行func函数
    t.start()

 　　我们可以看到，简单版的线程池就是简单的，逻辑就是每次执行一个任务就从队列取一个线程类创建一个线程，所以就有了--多少个任务，多少个线程，那和进程池相比较下，你会从中发现哪些不足呢？

• 第一，首先问你，执行完的线程去哪呢？--被程序回收销毁了！   那么执行完这些任务有必要一对一的创建线程吗？？--如果任务执行的快，一个线程有时候可以做多个任务
• 第二，上面只规定了队列的长度，并没有规定线程池里的线程数量？？
• 第三，进程池有回调函数这么一说法，上面没有？
• 第四，进程池里提供了close和terminate方法...

 

绝版

　　

　　好！我们来看看代码，是怎样解决上面不足的：

import queue
import threading
import contextlib
import time

#放入队列里，做为线程停止运行的信号
StopEvent = object()


class ThreadPool(object):

    def __init__(self, max_num, max_task_num = None):
        #对max_task_num进行判断
        if max_task_num:
            #如果有值传入，在创建队列时，就按传入值限定队列长度
            self.q = queue.Queue(max_task_num)
        else:
            #否则，就默认为队列长度为无限长
            self.q = queue.Queue()
        #最大线程数
        self.max_num = max_num
        #close方法的状态码
        self.cancel = False
        #terminate方法的状态码
        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
        """

        #如果执行close方法，这里就会执行return--中止函数，等同不再创建线程
        if self.cancel:
            return
        #如果空闲线程列表里为空并且已生成的线程数没有超过规定的最大线程数
        if len(self.free_list) == 0 and len(self.generate_list) < self.max_num:
            #那么就调用generate_thread方法创建线程
            self.generate_thread()
        #打包任务
        w = (func, args, callback,)
        #把打包的任务放入到队列里
        self.q.put(w)

    def generate_thread(self):

        #创建了一个线程
        t = threading.Thread(target=self.call)
        #启动线程，执行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执行结果
                result = func(*arguments)
                #并给执行状态赋值True
                success = True
            except Exception as e:
                #执行失败，赋值执行状态False
                success = False
                #同时赋值执行结果为None
                result = None

            if callback is not None:
                #如果callback不是默认None，就尝试下列操作
                try:
                    #把执行状态和执行结果传给回调函数，执行
                    callback(success, result)
                except Exception as e:
                    pass

            #with上下文管理--等同把free_list,current_thread传给worker_state方法，并执行
            with self.worker_state(self.free_list, current_thread):
                #判断self.terminal的状态
                if self.terminal:
                    #如果状态为True，就把任务变量设置为停止信号，等同于中止当前线程
                    event = StopEvent
                else:
                    #否则，就去队列里取任务
                    event = self.q.get()
        else:
            #如果是停止信号，就把当前线程从生成线程列表中移除
            self.generate_list.remove(current_thread)

    def close(self):
        """
        执行完所有的任务后，所有线程停止
        """

        #当执行close方法时，就把cancel状态设置为True
        self.cancel = True
        full_size = len(self.generate_list)
        while full_size:
            #往队列里加停止信号，直到生成线程列表长度为0
            self.q.put(StopEvent)
            full_size -= 1

    def terminate(self):
        """
        无论是否还有任务，终止线程
        """

        #当执行terminate方法时，把terminal状态设置为True
        self.terminal = True
        #因为突然中止，难免队列里还有任务，所以先清空一下队列
        self.q.empty()
        while self.generate_list:
            #往队列里加停止信号，值到生成线程列表为空时
            self.q.put(StopEvent)



    @contextlib.contextmanager
    def worker_state(self, state_list, worker_thread):
        """
        用于记录线程中正在等待的线程数
        """

        #把执行完任务的当前线程添加到空闲线程列表里
        state_list.append(worker_thread)
        try:
            #暂时跳出
            yield
        finally:
            #把当前线程从空闲线程列表里移除
            state_list.remove(worker_thread)



# How to use


pool = ThreadPool(5)

def callback(status, result):
    # status, execute action status
    # result, execute action return value
    pass


def action(i):
    print(i)

for i in range(30):
    ret = pool.run(action, (i,), callback)

time.sleep(5)
print(len(pool.generate_list), len(pool.free_list))
print(len(pool.generate_list), len(pool.free_list))
# pool.close()
# pool.terminate()

           欢迎大家对我的博客内容提出质疑和提问！谢谢

　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　                  笔者：拍省先生