From 32266eaaa8057a7f351ff2ee71cb12ca6eabaa59 Mon Sep 17 00:00:00 2001 From: ForgQi <34411314+ForgQi@users.noreply.github.com> Date: Sat, 26 Feb 2022 22:08:48 +0800 Subject: [PATCH] Add named thread pool --- biliup/engine/event.py | 19 +++++++++---------- biliup/handler.py | 6 +++--- 2 files changed, 12 insertions(+), 13 deletions(-) diff --git a/biliup/engine/event.py b/biliup/engine/event.py index a89cd86..3569580 100644 --- a/biliup/engine/event.py +++ b/biliup/engine/event.py @@ -21,9 +21,10 @@ class EventManager(Thread): # 事件管理器开关 self.__active = True # 事件处理线程池1 - self._pool1 = ThreadPoolExecutor(pool1_size, thread_name_prefix='Asynchronous1') - # 事件处理线程池2 - self._pool2 = ThreadPoolExecutor(pool2_size, thread_name_prefix='Asynchronous2') + self._pool = { + 'Asynchronous1': ThreadPoolExecutor(pool1_size, thread_name_prefix='Asynchronous1'), + 'Asynchronous2': ThreadPoolExecutor(pool2_size, thread_name_prefix='Asynchronous2') + } # 阻塞函数列表 self.__block = [] @@ -47,10 +48,7 @@ class EventManager(Thread): # 若存在,则按顺序将事件传递给处理函数执行 for handler in self.__handlers[event.type_]: if handler.__qualname__ in self.__block: - if event.type_ == 'download': - self._pool1.submit(handler, event) - else: - self._pool2.submit(handler, event) + self._pool.get(handler.pool).submit(handler, event) else: handler(event) @@ -59,8 +57,8 @@ class EventManager(Thread): # 将事件管理器设为停止 self.__active = False self.__eventQueue.put(None) - self._pool1.shutdown() - self._pool2.shutdown() + for pool in self._pool.values(): + pool.shutdown() def add_event_listener(self, type_, handler): """绑定事件和监听器处理函数""" @@ -122,7 +120,7 @@ class EventManager(Thread): return _event self.add_event_listener(type_, wrapper) - + wrapper.pool = block return wrapper else: def decorator(func): @@ -137,6 +135,7 @@ class EventManager(Thread): callback(_event) return _event + wrapper.pool = block return wrapper return decorator diff --git a/biliup/handler.py b/biliup/handler.py index e745b45..f133914 100644 --- a/biliup/handler.py +++ b/biliup/handler.py @@ -36,7 +36,7 @@ def create_event_manager(): event_manager = create_event_manager() -@event_manager.register(DOWNLOAD, block=True) +@event_manager.register(DOWNLOAD, block='Asynchronous1') def process(name, url): date = common.time_now(config['streamers'][name].get('title')) try: @@ -50,7 +50,7 @@ def process(name, url): return Event(UPLOAD, (name, url, date)) -@event_manager.register(UPLOAD, block=True) +@event_manager.register(UPLOAD, block='Asynchronous2') def process_upload(name, url, date): yield Event(BE_MODIFIED, (url, 2)) try: @@ -72,7 +72,7 @@ class KernelFunc: self.inverted_index = inverted_index self.streamer_url = streamer_url - @event_manager.register(CHECK, block=True) + @event_manager.register(CHECK, block='Asynchronous1') def singleton_check(self, platform): plugin = self.checker[platform] wait = config.get('checker_sleep') if config.get('checker_sleep') else 15