# 核心目的是提供异步任务执行、回调处理和任务状态管理功能
from concurrent.futures import ThreadPoolExecutor,as_completed
import time
import threading
from types import MethodType
class ThreadTask:# 长期各类任务线程池def __init__(self):self.executor = ThreadPoolExecutor(max_workers=3)self.future = [] #活跃等待任务self.count = 0 #任务计数且充当iddef run_call_back(self,fn):# 检查 Future 是否被取消if not fn.cancelled():self.future_call_back(fn)else:print('任务已被取消')def future_call_back(self, fn):print(f"回调结果:{fn.result()}")def add_task(self,fun,*args,**kwargs):future = self.executor.submit(fun,*args,**kwargs)self.count += 1self.update_future()task_info = {'future': future,'name': fun.__name__,'args': args,'kwargs': kwargs,'task_id': self.count,'submit_time': time.ctime()}self.future.append(task_info)# 当任务结束时进行回调，相比于在外部获取result()，可以不阻塞且自动处理结果future.add_done_callback(self.run_call_back)return self.count# 等待所有任务完成def wait_all_result(self):for future in as_completed(self.future):future.result()# 取消某个任务def cancel_future(self, task_id = None):self.update_future()cancel_task = list(filter(lambda x:x['task_id'] == task_id,self.future))if cancel_task:if cancel_task[0]['future']._state == 'PENDING':cancel_task[0]['future'].cancel()print('取消任务')else:print('任务已启动，无法取消')else:print('任务为不活跃状态，无需取消') # 获取当前任务信息@propertydef get_task_info(self):self.update_future()info = {'all_task_count': self.count,'now_active_task_info': self.future}return infodef update_future(self):self.future = list(filter(lambda x:x['future']._state in ['RUNNING', 'PENDING'], self.future))def shutdown(self):self.executor.shutdown()
def task(name, duration, par = 'test'):print(f"{time.ctime()}任务 {name} 开始执行",par)time.sleep(duration)print(f"任务 {name} 完成",par)return f"结果_{name}"
thread = ThreadTask()
def future_call_back(self, fn):print(f"特别回调结果:{fn.result()}")
# 如果某个实例需要额外的回调方法，修改回调方法
thread.future_call_back = MethodType(future_call_back,thread)
task_id = []
for i in range(6):submit_task_id = thread.add_task(task,'user'+str(i),i,par='test'+str(i))task_id.append(submit_task_id)
thread.cancel_future(task_id[-1])
print(thread.future)
time.sleep(1)
thread.cancel_future(task_id[0])
thread.cancel_future(task_id[-3])
print('当前任务',thread.get_task_info)
