python第十一天-----补：线程池

低版本：

#!/usr/bin/env python
import threading
import time
import queue


class TreadPool:
    """
    将线程加入到队列中作为资源去完成任务
    优点：简单好写容易理解
    缺点：太尼玛多了.....
    """
    def __init__(self, maxsize):
        self.maxsize = maxsize
        self._q = queue.Queue(maxsize)
        for i in range(maxsize):
            self._q.put(threading.Thread)

    def get_thread(self):
        return self._q.get()

    def add_thread(self):
        self._q.put(threading.Thread)


def task(arg, p):
    print(arg)
    time.sleep(1)
    p.add_thread()

pool = TreadPool(5)

for i in range(100):
    t = pool.get_thread()
    obj = t(target=task, args=(i, pool))
    obj.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, max_task_num=None):
        """
        构造方法
        :param max_num:
        :param max_task_num:所创建的队列内最大支持的任务个数
        """
        if max_task_num:
            self.q = queue.Queue(max_task_num)  # 指定队列任务数量则创建有限队列
        else:
            self.q = queue.Queue()              # 未指定队列任务数量则创建无限队列
        self.max_num = max_num                  # 每次使用的最大线程个数
        self.cancel = False                     # 任务取消，默认False,用于线程停止的判断
        self.terminal = False                   # 任务终止，默认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 self.cancel:                         # 如果条件为真则不会继续执行
            return
        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)  # 调用线程类创建一个线程，参数传递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)               # 结果为任务处理的出的结果
                success = True                          # 任务处理成功
            except Exception as e:
                success = False                         # 任务处理失败
                result = None                           # 结果为None

            if callback is not None:                    # 回调不为空
                try:
                    callback(success, result)            # 将刚才执行结果返回
                except Exception as e:
                    pass

            with self.worker_state(self.free_list, current_thread):
                if self.terminal:                       # 如果线程池已经被终止
                    event = StopEvent                   # 事件变为空任务
                else:   
                    event = self.q.get()                # 事件为正常任务
        else:

            self.generate_list.remove(current_thread)   # 生成任务列表移除当前任务

    def close(self):
        """
        执行完所有的任务后，所有线程停止
        """
        self.cancel = True                      # 线程停止，判定条件变为真
        full_size = len(self.generate_list)     # 获取还有几个在执行任务的线程
        while full_size:                        # 向队列中添加相应个数的空任务
            self.q.put(StopEvent)
            full_size -= 1

    def terminate(self):
        """
        无论是否还有任务，终止线程
        """
        self.terminal = True                    # 线程池关闭，判定条件变为真

        while self.generate_list:               # 当还有线程存在时放置空任务
            self.q.put(StopEvent)

        self.q.queue.clear()                    # 将队列中所有任务清空

    @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)                            # 创建一个每次支持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):                                     # 使用线程池执行30次任务
    ret = pool.run(action, (i,), callback)

time.sleep(1)                                           # 1秒等待
print(len(pool.generate_list), len(pool.free_list))     # 打印线程池内当前任务个数及空任务个数
pool.close()                                            # 线程停止
pool.terminate()                                        # 线程池终止