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

Posted 2020-08-11

低版本：

 1 #!/usr/bin/env python
 2 import threading
 3 import time
 4 import queue
 5 
 6 
 7 class TreadPool:
 8     """
 9     将线程加入到队列中作为资源去完成任务
10     优点：简单好写容易理解
11     缺点：太尼玛多了.....
12     """
13     def __init__(self, maxsize):
14         self.maxsize = maxsize
15         self._q = queue.Queue(maxsize)
16         for i in range(maxsize):
17             self._q.put(threading.Thread)
18 
19     def get_thread(self):
20         return self._q.get()
21 
22     def add_thread(self):
23         self._q.put(threading.Thread)
24 
25 
26 def task(arg, p):
27     print(arg)
28     time.sleep(1)
29     p.add_thread()
30 
31 pool = TreadPool(5)
32 
33 for i in range(100):
34     t = pool.get_thread()
35     obj = t(target=task, args=(i, pool))
36     obj.start()

高级版本：

 

  1 #!/usr/bin/env python
  2 # -*- coding:utf-8 -*-
  3 
  4 import queue        # 队列模块
  5 import threading    # 线程模块
  6 import contextlib   # 上下文模块
  7 import time         # 时间模块
  8 
  9 StopEvent = object()            # 创建一个停止时所需要用到的对象
 10 
 11 
 12 class ThreadPool(object):
 13     """
 14     线程池（用于放置任务，将任务作为队列中元素让线程去取得，可以复用线程减少开销）
 15     """
 16     def __init__(self, max_num, max_task_num=None):
 17         """
 18         构造方法
 19         :param max_num:
 20         :param max_task_num:所创建的队列内最大支持的任务个数
 21         """
 22         if max_task_num:
 23             self.q = queue.Queue(max_task_num)  # 指定队列任务数量则创建有限队列
 24         else:
 25             self.q = queue.Queue()              # 未指定队列任务数量则创建无限队列
 26         self.max_num = max_num                  # 每次使用的最大线程个数
 27         self.cancel = False                     # 任务取消，默认False,用于线程停止的判断
 28         self.terminal = False                   # 任务终止，默认False,用于线程池终止的判断
 29         self.generate_list = []                 # 定义一个已生成任务列表
 30         self.free_list = []                     # 定义一个空闲任务列表
 31 
 32     def run(self, func, args, callback=None):
 33         """
 34         线程池执行一个任务方法
 35         :param func: 传递进来的任务函数
 36         :param args: 任务函数使用的参数
 37         :param callback: 任务执行失败或成功后执行的回调函数，回调函数有两个参数
 38                             1、任务函数执行状态；
 39                             2、任务函数返回值（默认为None，即：不执行回调函数）
 40         :return: 如果线程池已经终止，则返回True否则None
 41         """
 42         if self.cancel:                         # 如果条件为真则不会继续执行
 43             return
 44         if len(self.free_list) == 0 and len(self.generate_list) < self.max_num:
 45             self.generate_thread()              # 如果现有空闲列表无元素并且已生成任务列表内
 46                                                 # 元素个数小于队列支持的最大数量则创建一个线程
 47         w = (func, args, callback,)             # 具体任务
 48         self.q.put(w)                           # 将任务放入队列当中
 49 
 50     def generate_thread(self):
 51         """
 52         创建一个线程方法
 53         """
 54         t = threading.Thread(target=self.call)  # 调用线程类创建一个线程，参数传递self.call方法
 55         t.start()                               # 线程开始
 56 
 57     def call(self):
 58         """
 59         循环去获取任务函数并执行任务函数
 60         """
 61         current_thread = threading.currentThread()      # 创建当前任务
 62         self.generate_list.append(current_thread)       # 生成任务列表添加当前任务
 63 
 64         event = self.q.get()                            # 事件获取
 65         while event != StopEvent:                       # 当前事件不是停止时执行
 66 
 67             func, arguments, callback = event           # 任务具体函数，参数获取
 68             try:
 69                 result = func(*arguments)               # 结果为任务处理的出的结果
 70                 success = True                          # 任务处理成功
 71             except Exception as e:
 72                 success = False                         # 任务处理失败
 73                 result = None                           # 结果为None
 74 
 75             if callback is not None:                    # 回调不为空
 76                 try:
 77                     callback(success, result)            # 将刚才执行结果返回
 78                 except Exception as e:
 79                     pass
 80 
 81             with self.worker_state(self.free_list, current_thread):
 82                 if self.terminal:                       # 如果线程池已经被终止
 83                     event = StopEvent                   # 事件变为空任务
 84                 else:   
 85                     event = self.q.get()                # 事件为正常任务
 86         else:
 87 
 88             self.generate_list.remove(current_thread)   # 生成任务列表移除当前任务
 89 
 90     def close(self):
 91         """
 92         执行完所有的任务后，所有线程停止
 93         """
 94         self.cancel = True                      # 线程停止，判定条件变为真
 95         full_size = len(self.generate_list)     # 获取还有几个在执行任务的线程
 96         while full_size:                        # 向队列中添加相应个数的空任务
 97             self.q.put(StopEvent)
 98             full_size -= 1
 99 
100     def terminate(self):
101         """
102         无论是否还有任务，终止线程
103         """
104         self.terminal = True                    # 线程池关闭，判定条件变为真
105 
106         while self.generate_list:               # 当还有线程存在时放置空任务
107             self.q.put(StopEvent)
108 
109         self.q.queue.clear()                    # 将队列中所有任务清空
110 
111     @contextlib.contextmanager
112     def worker_state(self, state_list, worker_thread):
113         """
114         用于记录线程中正在等待的线程数
115         """
116         state_list.append(worker_thread)        # 等待状态列表中添加正在等待的线程数
117         try:
118             yield
119         finally:
120             state_list.remove(worker_thread)    # 移除正在等待的线程数
121 
122 
123 
124 # How to use
125 
126 
127 pool = ThreadPool(5)                            # 创建一个每次支持5线程的线程池
128 
129 def callback(status, result):
130     # status, execute action status
131     # result, execute action return value
132     pass
133 
134 
135 def action(i):                                          # 任务函数
136     print(i)
137 
138 for i in range(30):                                     # 使用线程池执行30次任务
139     ret = pool.run(action, (i,), callback)
140 
141 time.sleep(1)                                           # 1秒等待
142 print(len(pool.generate_list), len(pool.free_list))     # 打印线程池内当前任务个数及空任务个数
143 pool.close()                                            # 线程停止
144 pool.terminate()                                        # 线程池终止