Python 執行緒池實現(轉)
阿新 • • 發佈:2021-06-18
轉自武Sir的部落格 https://www.cnblogs.com/wupeiqi/articles/4839959.html
import queue import threading import contextlib import time StopEvent = object() class ThreadPool(object): def __init__(self, max_num, max_task_num = None): if max_task_num: self.q = queue.Queue(max_task_num) else: self.q = queue.Queue() self.max_num = max_num self.cancel = False 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 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) 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 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) 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()