Merge pull request #222 from LINSTCL/threadpool-optimization

使用线程池控制线程数量，防止高并发崩溃

config-template.py

@@ -202,6 +202,11 @@ hide_exce_info_to_user = False
 # 设置为空字符串时，不发送提示信息
 alter_tip_message = '出错了，请稍后再试'
 
+# 机器人线程池大小
+# 该参数决定机器人可以同时处理几个人的消息，超出线程池数量的请求会被阻塞，不会被丢弃
+# 如果你不清楚该参数的意义，请不要更改
+pool_num = 10
+
 # 每个会话的过期时间，单位为秒
 # 默认值20分钟
 session_expire_time = 60 * 20

main.py

@@ -182,7 +182,8 @@ def main(first_time_init=False):
         # 初始化qq机器人
         qqbot = pkg.qqbot.manager.QQBotManager(mirai_http_api_config=config.mirai_http_api_config,
                                                timeout=config.process_message_timeout, retry=config.retry_times,
-                                               first_time_init=first_time_init)
+                                               first_time_init=first_time_init,
+                                               pool_num=config.pool_num if hasattr(config, 'pool_num') else 10)
 
         # 加载插件
         import pkg.plugin.host

pkg/qqbot/manager.py

@@ -2,6 +2,7 @@ import asyncio
 import json
 import os
 import threading
+from concurrent.futures import ThreadPoolExecutor
 
 import mirai.models.bus
 from mirai import At, GroupMessage, MessageEvent, Mirai, StrangerMessage, WebSocketAdapter, HTTPAdapter, \
@@ -20,13 +21,6 @@ import pkg.utils.context
 import pkg.plugin.host as plugin_host
 import pkg.plugin.models as plugin_models
 
-
-# 并行运行
-def go(func, args=()):
-    thread = threading.Thread(target=func, args=args, daemon=True)
-    thread.start()
-
-
 # 检查消息是否符合泛响应匹配机制
 def check_response_rule(text: str):
     config = pkg.utils.context.get_config()
@@ -55,6 +49,9 @@ def check_response_rule(text: str):
 class QQBotManager:
     retry = 3
 
+    #线程池控制
+    pool = None
+
     bot: Mirai = None
 
     reply_filter = None
@@ -64,11 +61,14 @@ class QQBotManager:
     ban_person = []
     ban_group = []
 
-    def __init__(self, mirai_http_api_config: dict, timeout: int = 60, retry: int = 3, first_time_init=True):
-
+    def __init__(self, mirai_http_api_config: dict, timeout: int = 60, retry: int = 3, pool_num: int = 10, first_time_init=True):
         self.timeout = timeout
         self.retry = retry
 
+        self.pool_num = pool_num
+        self.pool = ThreadPoolExecutor(max_workers=self.pool_num)
+        logging.debug("Registered thread pool Size:{}".format(pool_num))
+
         # 加载禁用列表
         if os.path.exists("banlist.py"):
             import banlist
@@ -116,7 +116,7 @@ class QQBotManager:
 
                 self.on_person_message(event)
 
-            go(friend_message_handler, (event,))
+            self.go(friend_message_handler, event)
 
         @self.bot.on(StrangerMessage)
         async def on_stranger_message(event: StrangerMessage):
@@ -136,7 +136,7 @@ class QQBotManager:
 
                 self.on_person_message(event)
 
-            go(stranger_message_handler, (event,))
+            self.go(stranger_message_handler, event)
 
         @self.bot.on(GroupMessage)
         async def on_group_message(event: GroupMessage):
@@ -156,7 +156,7 @@ class QQBotManager:
 
                 self.on_group_message(event)
 
-            go(group_message_handler, (event,))
+            self.go(group_message_handler, event)
 
         def unsubscribe_all():
             """取消所有订阅
@@ -173,6 +173,9 @@ class QQBotManager:
 
         self.unsubscribe_all = unsubscribe_all
 
+    def go(self, func, *args, **kwargs):
+        self.pool.submit(func, *args, **kwargs)
+
     def first_time_init(self, mirai_http_api_config: dict):
         """热重载后不再运行此函数"""