# parallel模块说明 模块名：HiveNetCore.parallel parallel是并行任务（多线程、多进程）的处理框架和实现，包括并行任务处理，以及并行池（类似线程池或进程池）管理的封装，同时该框架可以支持扩展为分布式的并行任务处理（按框架定义实现对应的实现类即可）。 使用该框架的最主要优点为： 1、多线程、多进程接口统一，可直接实现多进程或多线程的直接切换（无需修改代码）； 2、简化多线程和多进程的代码实现，且实现了函数执行回调函数的支持；此外遵循HiveNet 日志规范（hivenet_log_standards）； 3、框架可支持实现分布式计算并行任务的自定义扩展。 ## ParallelLockFw（并行锁框架） 定义了并行任务所需的锁对象框架，该框架与Python自带的threading.RLock定义兼容，目前已直接利用Python原生类实现了该框架的线程锁实例（ThreadParallelLock）、进程锁实例（ProcessParallelLock），分别可以针对多线程、多进程的锁控制。后续也可根据需要基于框架（ParallelLockFw）实现分布计算锁，来支持分布式计算的锁控制需求。 使用ParallelLockFw具体实例的简单示例如下： ``` ``` ## ParallelShareDictFw(并行任务共享字典框架) 定义了并行任务所需的数据共享框架（特别是多进程、或后续可能的分布式任务的数据共享），目前已实现了线程共享字典实例（ThreadParallelShareDict，利用全局变量共享）、进程共享字典实例（ProcessParallelShareDict，利用multiprocessing的Manager对象共享）。后续也可以根据需要基于框架（ParallelShareDictFw）实现分布共享字典，来支持分布式计算的共享数据需求。 使用ParallelShareDictFw具体实例的简单示例如下： ``` # 创建进程共享字典 _tag = uuid.uuid1() # 共享字典标签 _dict = ProcessParallelShareDict(_tag) # 使用共享字典 _dict['key'] += 1 ``` ## ParallelFw(并行任务框架) 定义了并行任务处理的框架（例如多线程、多进程的任务发起），目前已实现了多线程任务实例（ThreadParallel）、多进程任务实例（ProcessParallel）。后续也可以根据需要基于框架（ParallelFw）实现分布式任务处理实例，来支持分布式计算的并行任务处理需求。 支持的方法包括： start - 启动并行池任务处理 stop - 停止并行池任务处理 pause - 暂停并行池任务处理 resume - 恢复并行池任务处理 并行任务框架的应用的关键在于对象初始化，支持的参数如下： ``` @param {fuction} deal_fun - 并发处理主函数，按顺序入参，可以有返回值 @param {tuple} run_args=None - 并发处理主函数的入参列表 @param {dict} run_kwargs=None - 并发处理主函数的动态入参列表 @param {bool} auto_start=False - 是否创建时自动发起并发任务 @param {string} pid='' - 并发对象的id @param {string} pname='' - 并发对象的标识名 @param {ParallelLockFw} lock=None - 并发锁对象，控制多个并发对象根据该锁控制单一处理 @param {fuction} callback_fun=None - 回调函数，在执行完函数后执行，定义如下 fun(id, name, call_result, deal_fun_ret) id - 并发对象的id name - 并发对象的标识名 call_result - 线程执行的结果，CResult对象，如果执行无异常返回'00000'； 如果发生异常，返回'21399'，并登记异常信息 deal_fun_ret - deal_fun函数执行的返回值 @param {bool} set_daemon=False - 是否设置守护，如果设置守护，则在主进程中使用join方法等待所有并发完成， 否则主进程结束后并发执行的结果不可预知；如果不设置守护，主进程结束后并发任务仍会执行 @param {Logger} logger=None - 日志对象，如果为None代表不需要输出日志，传入对象需满足: 1、标准logging的logger对象 2、自定义的日志类对象，但应实现info、warning、error等标准方法 @param {bool} is_use_global_logger=True - 当logger=None时，是否使用全局logger对象 注：通过RunTool.set_global_logger进行设置 @param {int} log_level=logging.INFO - 打印日志的级别 @param {bool} use_distributed_logger=False - 是否使用分布式logger，如果是，则每个分布任务自行创建独立logger记录日志 注：对于多进程及分布式并发任务，应采取该日志模式 @param {string} distributed_logger_module_name='' - 分布式日志类模块名 @param {string} distributed_logger_class_name='' - 分布式日志类类名 @param {tuple} distributed_logger_args=None - 分布式日志类创建参数 @param {dict} distributed_logger_kwargs=None - 分布式日志类创建参数 @param {dict} distributed_logger_replace_para=dict() - 分布式日志动态参数替换，在每创建一个并发任务通过该参数修改日志参数 key - 如果是int类型，代表替换distributed_logger_args的第几个参数；如果是string，代表替换distributed_logger_kwargs的指定参数 value - 替换参数字典，key为要替换正则表达式字符，value为要替换的动态值 要替换的动态值，可选值有以下几种： 'pid' - 并发任务id 'pname' - 并发任务名 'pocess_id' - 进程ID 'thread_id' - 线程ID - 暂不支持 @param {bool} is_logger_to_deal_fun=False - 是否传递并发任务logger到deal_fun中 注意：传递通过kwargs，参数名为logger ``` 该框架的几个特性说明如下： 1、并发任务处理函数（deal_fun）支持返回值，返回值（或执行过程的异常信息）可在处理函数执行完成后，通过回调函数（callback_fun）通知回调用方； 2、可在定义中直接传入并发任务锁（lock），框架会保证同一锁发起的并发任务基于该所控制并发； 3、并发任务的日志记录支持单一日志对象（多进程情况不支持）及分布式日志对象（每个任务自行初始化一个新的日志对象），可通过distributed_logger相关参数实现各种情况的日志输出支持；此外也可支持将日志对象传递到执行函数中（is_logger_to_deal_fun参数），执行函数可以复用该日志对象进行日志的输出； 使用ParallelFw具体实例的简单示例如下： ``` # 定义主进程使用的logger _logger = HiveNetCore.logging_hivenet.Logger(conf_file_name=None, logger_name=HiveNetCore.logging_hivenet.EnumLoggerName.Console, config_type=HiveNetCore.logging_hivenet.EnumLoggerConfigType.JSON_STR) # 定义并发任务的logger创建参数 _logger_kwargs = { 'conf_file_name': None, 'logger_name': HiveNetCore.logging_hivenet.EnumLoggerName.Console, 'config_type': HiveNetCore.logging_hivenet.EnumLoggerConfigType.JSON_STR } # 定义并发任务处理函数 def demo_deal_fun(a, b, c, p1='', p2=''): print('demo_deal_fun print: a=%s, b=%s, c=%s, p1=%s, p2=%s' % ( str(a), str(b), str(c), str(p1), str(p2) )) time.sleep(0.1) _ret = ('demo_deal_fun:r1-%s' % (str(a)), 'demo_deal_fun:r2') print('demo_deal_fun print: return=%s' % (str(_ret))) return _ret # 定义并发任务回调函数 def demo_callback(id, name, call_result, deal_fun_ret): print('demo_callback print: id=%s, name=%s, call_result=%s, deal_fun_ret=%s' % ( str(id), str(name), str(call_result), str(deal_fun_ret) )) return # 初始化并发任务对象 _t1 = ProcessParallel( deal_fun=demo_deal_fun, run_args=('demo2-a-1', 'demo2-b-1', 'demo2-c-1'), run_kwargs={'p1': 'demo2-p1-1', 'p2': 'demo2-p2-1'}, pid='demo2-t1', pname='demo2-t1-name', callback_fun=demo_callback, set_daemon=True, logger=_logger, use_distributed_logger=True, distributed_logger_module_name='HiveNetCore.logging_hivenet', distributed_logger_class_name='Logger', distributed_logger_args=None, distributed_logger_kwargs=_logger_kwargs ) # 执行并发任务 _t1.start() # 等待任务完成 _t1.join() # 如果过程中想强行中止并发任务 _t1.force_stop() ``` ## ParallelPool（并行池管理类） 实现了统一的并行池（线程池、进程池的管理），实现池中任务的创建、销毁、启动、暂停等管理，同时直接支持队列方式的消费者任务处理。该类可以直接支持后续扩展其他并行任务的管理（如分布式并行任务，只要并行任务基于以上的框架实现）。 并行池的应用的关键在于对象初始化，支持的参数如下： ``` @param {fuction} deal_fun - 并发任务处理主函数，按顺序入参，可以有返回值 注：该函数内部需自行实现获取数据并处理的流程，但约定如果无处理数据，函数应返回None（用于并发池判断是否释放任务） @param {ParallelFw} parallel_class=None - 并行任务类定义对象，获取方法如下： getattr(ImportTool.import_module('HiveNetCore.parallel'), 'ThreadParallel') @param {tuple} run_args=None - 并发任务处理主函数的入参列表 @param {dict} run_kwargs=None - 并发任务处理主函数的动态入参列表 @param {string} pname='' - 并发任务处理主函数的标识名 @param {ParallelLockFw} lock=None - 并发锁对象，控制多个并发对象根据该锁控制单一处理 @param {fuction} callback_fun=None - 回调函数，在执行完函数后执行，定义如下 fun(id, name, call_result, deal_fun_ret) id - 并发对象的id name - 并发对象的标识名 call_result - 线程执行的结果，CResult对象，如果执行无异常返回'00000'； 如果发生异常，返回'21399'，并登记异常信息 deal_fun_ret - deal_fun函数执行的返回值 @param {Logger} logger=None - 日志对象，如果为None代表不需要输出日志，传入对象需满足: 1、标准logging的logger对象 2、自定义的日志类对象，但应实现info、warning、error等标准方法 @param {bool} is_use_global_logger=True - 当logger=None时，是否使用全局logger对象 注：通过RunTool.set_global_logger进行设置 @param {int} log_level=logging.INFO - 打印日志的级别 @param {bool} use_distributed_logger=False - 是否使用分布式logger，如果是，则每个分布任务自行创建独立logger记录日志 注：对于多进程及分布式并发任务，应采取该日志模式 @param {string} distributed_logger_module_name='' - 分布式日志类模块名 @param {string} distributed_logger_class_name='' - 分布式日志类类名 @param {tuple} distributed_logger_args=None - 分布式日志类创建参数 @param {dict} distributed_logger_kwargs=None - 分布式日志类创建参数 @param {dict} distributed_logger_replace_para=dict() - 分布式日志动态参数替换，在每创建一个并发任务通过该参数修改日志参数 key - 如果是int类型，代表替换distributed_logger_args的第几个参数；如果是string，代表替换distributed_logger_kwargs的指定参数 value - 替换参数字典，key为要替换正则表达式字符，value为要替换的动态值 要替换的动态值，可选值有以下几种： 'pid' - 并发任务id 'pname' - 并发任务名 'pocess_id' - 进程ID 'thread_id' - 线程ID @param {bool} is_logger_to_deal_fun=False - 是否传递并发任务logger到deal_fun中 注意：传递通过kwargs，参数名为logger @param {bool} auto_start=False - 是否自动启动并发池 @param {bool} auto_stop=False - 是否自动关闭并发池（当任务都已全部完成处理） @param {QueueFw} task_queue=None - 并发池需要处理的任务队列 注：如果有指定队列，get_task_num_fun参数无效，则自动根据队列长度检查待处理任务 @param {function} get_task_num_fun=None - 获取待处理任务数量的函数 注：如果task_queue和get_task_num_fun均为None，则直接创建最大数量的线程数，且不释放空闲任务 @param {list} get_task_num_fun_args=None - 获取待处理任务数量的函数，的入参列表 @param {int} maxsize=10 - 并发池最大并发对象数 @param {int} minsize=0 - 并发池最小并发对象数 @param {number} worker_release_time=10 - 空闲工作并发对象释放时间，单位为秒, 0代表不控制空闲释放 @param {number} worker_overtime=0 - 正在执行的任务超时时间（秒），0代表不控制超时 注：对于超时执行任务，将视为失效任务，可以选择直接忽略或强制中止 @param {bool} force_kill_overtime_worker=False - 是否强制中止失效任务 @param {bool} replace_overtime_worker=False - 是否创建新任务替代超时任务 注：仅当force_kill_overtime_worker=False时才会进行替代 @param {number} daemon_thread_time=0.01 - 守护线程的间隔时间 @param {ParallelShareDictFw} sharedict_class=None - 进程间共享字典对象的类对象，获取方法如下： getattr(ImportTool.import_module('HiveNetCore.parallel'), 'ThreadParallelShareDict') @param {ParallelLockFw} parallel_lock_class=None - 进程间锁对象的类对象，获取方法如下： getattr(ImportTool.import_module('HiveNetCore.parallel'), 'ThreadParallelLock') ``` 该框架的几个特性说明如下： 1、支持多种并行任务类型，可通过参数指定具体的并行任务类型（parallel_class，注意要取到的是该类型的定义对象，通过getattr获取，后续基于该对象动态创建并行任务，例如多线程还是多进程）； 2、支持指定队列发起并行任务（task_queue，队列对象应为Python原生队列对象，或基于HiveNetCore.queue_hivenet.QueueFw扩展的队列对象）；如果需要执行固定数量的任务，可以将任务放入队列后，指定自动结束参数（auto_stop），再启动并发池，并发池将在任务完成后自行结束； 3、支持自动释放空闲并发任务（worker_release_time），减少并发任务空跑导致的性能消耗； 4、支持处理超时任务（worker_overtime），可强制中止超时任务（force_kill_overtime_worker），或启动新任务替代某一超时任务（replace_overtime_worker），避免超时任务堵塞了任务队列的处理； 使用ParallelPool的简单示例如下： ``` # 定义并行池的处理函数，注意示例中q为支持进程共享访问的队列 def _deal_fun_demo5(q, a, p1=''): try: obj = q.get(block=False) print('_deal_fun_demo5: a=%s, p1=%s, qobj=%s' % (str(a), str(p1), str(obj))) time.sleep(1) return True except Exception as e: print('_deal_fun_demo5 error: a=%s, p1=%s, error=%s' % (str(a), str(p1), str(type(e)))) time.sleep(1) return None def demo(): # 定义要处理的队列 _task_queue = multiprocessing.Queue() _i = 0 while _i < 100: _task_queue.put(_i) _i = _i + 1 # 定义进程池，注意大部分参数与ParallelFw一致 _pool = ParallelPool( deal_fun=_deal_fun_demo5, parallel_class=getattr(ImportTool.import_module('HiveNetCore.parallel'), 'ProcessParallel'), run_args=(_task_queue, 'demo5-a'), run_kwargs={'p1': 'demo5-p1'}, pname='demo5', callback_fun=demo_callback, logger=_logger, use_distributed_logger=True, distributed_logger_module_name='HiveNetCore.logging_hivenet', distributed_logger_class_name='Logger', distributed_logger_args=None, distributed_logger_kwargs=_logger_kwargs, auto_start=False, auto_stop=True, task_queue=_task_queue, get_task_num_fun=None, get_task_num_fun_args=None, maxsize=10, minsize=0, worker_release_time=10, worker_overtime=0, force_kill_overtime_worker=False, replace_overtime_worker=False, daemon_thread_time=0.01, sharedict_class=getattr(ImportTool.import_module('HiveNetCore.parallel'), 'ProcessParallelShareDict'), parallel_lock_class=getattr(ImportTool.import_module('HiveNetCore.parallel'), 'ProcessParallelLock') ) # 启动并行池，该示例定义了自动结束 _pool.start() while not _pool.is_stop: time.sleep(1) ```