Add named thread pool

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
 

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