import time
import sys, os
import threading
import gevent
from multiprocessing import Process, Pool, freeze_support
import gevent.pool as pool
from gevent import monkey
import typespool_key = ['id', 'func', 'args']# daemon            是否为守护进程(线程)
# is_wait_end       父进程是否等待任务结束
# is_sync           任务是否串行
# all_true          任务集中有任务出错立即结束全部  退出当前进程或进程池 结束协程
# Readme            为True打印帮助信息def printruntime(f):def run(*args, **kwargs):s = time.time()f(*args, **kwargs)print(time.time() - s)return rundef dis_erro(e, is_alltrue=False, id=None):if id != None:print('程序出错 id:' + str(id) + ' msg: ', e, )else:print('程序出错  msg: ', e, )if is_alltrue == True:print('已结束全部任务')#########################################################################
def creat_sub_thread(func, args, daemon=True, is_wait_end=False):try:st = threading.Thread(target=func, args=tuple(args), )if daemon == True:st.daemon = Truest.start()if is_wait_end == True:st.join()except Exception as e:dis_erro(e=e, )def creat_sub_process(func, args, daemon=True, is_wait_end=False):sp = Process(target=func, args=tuple(args))try:if daemon == True:sp.daemon = Truesp.start()if is_wait_end == True:sp.join()except Exception as e:sp.terminate()dis_erro(e=e, )
#########################################################################def creat_thread_pool(threadlist, thread_lock_count, daemon=True, is_sync=False, all_true=False, is_wait_end=True, Readme=False):#   适用于密集cpu并发#   threadlist >>#   [#       {'id':id,'func':func,'args':(args)}, ...#   ] , ...#def print_readme():print("""#   适用于密集cpu并发#   threadlist >>#   [#       {'id':id,'func':func,'args':(args)}, ...#   ] , ...#""")class thread_job:def __init__(self, threadlist, thread_lock_count, daemon=True, is_sync=False, all_true=False, is_wait_end=True):self.threadlist = threadlistself.thread_lock_count = thread_lock_countself.daemon = daemonself.is_sync = is_syncself.is_wait_end = is_wait_endself.all_true = all_truedef check(self):for one in [self.all_true, self.daemon, self.is_sync, self.is_wait_end, ]:if type(one) is not bool:print('all_true is_sync daemon is_wait_end must be bool')return 'erro'if type(self.thread_lock_count) is not int:print('thread_lock_count must be int')return 'erro'elif type(self.threadlist) is not list:print('threadlist must be list')return 'erro'for t in threadlist:if type(t) is not dict:print('threadlist item must be dict')return 'erro'keylist = [x for x in t.keys()]if len(keylist) != 3:print("threadlist item like {'id':id,'func':func,'args':(args)}")return 'erro'for pk in pool_key:if pk not in keylist:print("threadlist item like {'id':id,'func':func,'args':(args)}")return 'erro'if not isinstance(t['func'], types.FunctionType):print('threadlist item first value type must be FunctionType ')return 'erro'if type(t['args']) is not tuple:print('threadlist item second value type must be tuple ')return 'erro'def job(self, id, func, tuple_args, count_lock):count_lock.acquire()cmd = 'func('for x in range(len(tuple_args)):cmd += 'tuple_args[' + str(x) + '],'cmd += ')'try:exec(cmd)except Exception as e:if self.all_true == True:dis_erro(e=e, id=str(id), is_alltrue=True)os._exit(-1)return 'all_true_erro'else:dis_erro(e=e, id=str(id),)count_lock.release()def run(self):count_lock = threading.BoundedSemaphore(self.thread_lock_count)thread_set = []for thread in self.threadlist:a = threading.Thread(target=self.job, args=(thread['id'], thread['func'], thread['args'], count_lock), )if self.daemon == True:  # or self.all_true == True:a.setDaemon(True)a.start()if self.is_sync == True:a.join()else:thread_set.append(a)if (self.is_sync == False and self.is_wait_end == True):for t in thread_set:t.join()if Readme == False:thread_job(threadlist, thread_lock_count, daemon, is_sync, all_true, is_wait_end).run()else:print_readme()def decorate(id, func, args):if not isinstance(func, types.FunctionType):return id, Exceptiontry:cmd = 'func('for i in range(len(args)):cmd += "args[" + str(i) + "],"cmd += ')'exec(cmd)except Exception as e:return id, edef creat_process_pool(processlist, process_lock_count, is_sync=False, all_true=False, is_wait_end=False, Readme=False):#   注意:进程池不能与协程池同时使用！#   开销巨大，一般最大值为内核数#   processlist >>#   [#      {'id':id,'func':func,'args':(args)}, ...#   ] , ...##class process_pool:def __init__(self, processlist, process_lock_count,  is_sync=False, all_true=False, is_wait_end=False, Readme=False):self.processlist = processlistself.process_lock_count = process_lock_countself.daemon = Trueself.is_sync = is_syncself.all_true = all_trueself.is_wait_end = is_wait_endself.Readme = Readmedef printreadme(self):print(""" #   注意:进程池不能与协程池同时使用！#   开销巨大，一般最大值为内核数#   processlist >>#   [#      {'id':id,'func':func,'args':(args)}, ...#   ] , ...#""")def check(self):for one in [self.all_true, self.daemon, self.is_sync, self.is_wait_end, self.Readme]:if type(one) is not bool:print('all_true is_sync daemon Readme is_wait_end must be bool')return 'erro'if type(self.process_lock_count) is not int:print('arg process_lock_count type must be int ')return 'erro'if type(self.processlist) is not list:print('arg processlist type must be list ')return 'erro'for x in self.processlist:if type(x) is not dict:print("list processlist item must be dict")return 'erro'keylist = [x for x in x.keys()]if len(keylist) != 3:print(" processlist item  must be {'id': id, 'func': func, 'args': args }")return 'erro'for pk in pool_key:if pk not in keylist:print(" processlist item  must be {'id': id, 'func': func, 'args': args }")return 'erro'if not isinstance(x['func'], types.FunctionType):print("list processlist key \'func\' value  must be function")return 'erro'if type(x['args']) is not tuple:print("list processlist key \'args\' value must be tuple")return 'erro'def __cb__(self, args):if type(args) is tuple:if isinstance(args[1], Exception):if self.all_true == True:dis_erro(e=args[1],id=args[0],is_alltrue=True)self.processpool.terminate()else:dis_erro(e=args[1], id=args[0], )def creat(self):if self.check() == 'erro':return -1freeze_support()self.processpool = Pool(processes=self.process_lock_count)try:for process in self.processlist:if is_sync == False:if len(process) == 3:self.processpool.apply_async(func=decorate, args=(process['id'], process['func'], process['args']), callback=self.__cb__)else:print("list CoroutinesList item [func,args,] ")raise TypeErrorelse:if len(process) == 3:try:self.processpool.apply(func=process['func'], args=process['args'], )except Exception as e:if self.all_true == True:dis_erro(e=e, id=str(process['id']),is_alltrue=True)self.processpool.terminate()return 'all_true_erro'else:dis_erro(e=e,id = str(process['id']))else:print("list CoroutinesList item [func,args]")raise TypeErrorexcept Exception as e:dis_erro(e=e,is_alltrue=True)self.processpool.terminate()self.processpool.close()if is_wait_end == True:self.processpool.join()def run(self):if self.daemon == True:if self.Readme == True:self.printreadme()else:try:self.creat()except Exception as e:dis_erro(e=e,)else:print('进程池必须是守护进程')process_pool(processlist, process_lock_count, is_sync, all_true, is_wait_end, Readme).run()def creat_coroutines_pool(coroutinesList, coroutines_lock_count,all_true=False):#  协程报错位置在源coroutinesList中，无法设置all_true#  注意:进程池不能与协程池同时使用！#  适用于I/O密集任务不适用与cpu密集任务#  [  协程列表#     {'id':id,'func':func,'args':(args)}, ...#   ]#monkey.patch_all()class CoroutinesPool:readme = """#  协程报错位置在源coroutinesList中，无法设置all_true#  注意:进程池不能与协程池同时使用！#  适用于I/O密集任务不适用与cpu密集任务#  [  协程列表#     {'id':id,'func':func,'args':(args)}, ...#   ]"""def __init__(self, CoroutinesList, coroutines_lock_count,all_true):self.coroutinesList = CoroutinesListself.coroutines_lock_count = coroutines_lock_countself.all_true = all_truedef Readme(self):print(self.readme)def check(self):if type(self.coroutines_lock_count) is not int:print('arg coroutines_lock_count type must be int ')return 'erro'for x in self.coroutinesList:if type(x) is not dict:print("list CoroutinesList item must be dict")return 'erro'keylist = [k for k in x.keys()]if len(keylist) != 3:print("CoroutinesList item like {'id':id,'func':func,'args':(args)}")return 'erro'for pk in pool_key:if pk not in keylist:print("threadlist item like {'id':id,'func':func,'args':(args)}")return 'erro'if not isinstance(x['func'], types.FunctionType):print("list CoroutinesList item like {'id':id,'func':func,'args':(args)}")return 'erro'if type(x['args']) is not tuple:print("list CoroutinesList item like {'id':id,'func':func,'args':(args)}")return 'erro'def decorate(self, id, func, args):# print(id,func,args,)cmd = 'func('for x in range(len(args)):cmd += 'args[' + str(x) + '],'cmd += ')'try:exec(cmd)except Exception as e:if self.all_true == True:dis_erro(e=e, id=str(id),is_alltrue=True )sys.exit()else:dis_erro(e=e, id=str(id), )def run(self):if self.check() == 'erro':return -1coroutines_list = []coroutinespool = pool.Pool(self.coroutines_lock_count)  # 协程池for coroutines in self.coroutinesList:try:coroutines_list.append(coroutinespool.spawn(self.decorate, coroutines['id'], coroutines['func'], coroutines['args']))except Exception as e:dis_erro(e=e, id=str( coroutines['id']))gevent.joinall(coroutines_list)try:CoroutinesPool(CoroutinesList=coroutinesList, coroutines_lock_count=coroutines_lock_count,all_true=all_true ).run()except Exception as e:print("run_CoroutinesPool erro ", e)###################################################################################if __name__ == '__main__':def p(*args):for i in args:time.sleep(1)print(i)return '123'def p1(*args):time.sleep(0.1)return 1 / 0def w(*args):with open('w.txt', 'w+') as f:f.write([x for x in args])def t(*args):time.sleep(5)job1 = [{'id': i, 'func': p, 'args': (1, 2, 3)} for i in range(5)]job2 = [{'id': i, 'func': p, 'args': (1, 2, 3)} for i in range(5)] + [{'id': 'erro', 'func': p1, 'args': (1, 2, 3)}] + [{'id': i, 'func': p, 'args': (1, 2, 3)} for i in range(5)]jobw = [{'id': 'w+', 'func': w, 'args': ('123',)}]jobtime = [{'id': 'time', 'func': t, 'args': ('123',)}]def test_t_p():creat_sub_process(func=creat_thread_pool, args=(job2, 10, True, False, False, False), is_wait_end=True, daemon=False)creat_thread_pool(job2, 4, daemon=False, is_sync=False, all_true=True,is_wait_end= False)creat_process_pool(job2, 3, is_sync=False, all_true=False, is_wait_end=True)creat_coroutines_pool(job2,5,all_true=True)creat_sub_process(func=creat_coroutines_pool,args=(job2,5,True),is_wait_end=False,daemon=False)print('==================')test_t_p()