refactor(perf): 建立三级资源防线并优化并发限流配置

- 【异常处理优化】引入自定义任务重复异常，精准区分“因重复运行而跳过”与“实际分析取消”两种不同语义，提升调度健壮性。 - 【三级并发控制架构】针对不同层级的资源消耗，建立了完整的限流保护体系： - 入口层：控制同时处于活跃状态的任务总数，防止瞬时加载大量消息导致服务器内存溢出 (OOM)； - 接口层：全局限制 LLM API 并发请求数，确保严格遵守服务商的频率限制 (RPM)； - 渲染层：全局限制浏览器图片生成进程，保护物理 CPU 与内存不被高能耗渲染任务耗尽。 - 【配置体验升级】新增「并发限流设置」配置组，统一收纳核心性能参数，并提供面向普通用户的防坑说明与更稳健的默认值。 - 【代码清理】移除陈旧的兼容逻辑与冗余参数，统一配置命名规范。 - 【功能精化】重排增量分析逻辑，确保在任务入口与分段处理流程中均有可靠的信号量保护。 - 【细节完善】优化群文件及群相册上传设置的文案描述，明确功能支持范围与权限要求。
2026-09-22 13:38:43 +00:00 · 2026-03-03 21:07:51 +08:00
parent ee0bf037b8
commit 7850697f1a
6 changed files with 111 additions and 56 deletions
@@ -112,12 +112,6 @@
                "items": {
                    "type": "string"
                }
-            },
-            "max_concurrent_tasks": {
-                "type": "int",
-                "description": "自动分析最大并发数",
-                "default": 1,
-                "hint": "同时进行的群聊分析任务数量，建议根据机器性能和服务商情况调整，过高可能导致LLM API RPM 超出限制，卡顿或被风控"
            }
        }
    },
@@ -406,5 +400,30 @@
                }
            }
        }
+    },
+    "performance": {
+        "description": "并发限流设置",
+        "type": "object",
+        "hint": "控制插件运行时的各项并发强度。合理的配置可以保护您的服务器不宕机、API 不被封禁。",
+        "items": {
+            "max_concurrent_groups": {
+                "type": "int",
+                "description": "最大活跃任务数（任务总闸）",
+                "default": 2,
+                "hint": "限制同时有多少个分析任务正在进行。主要用于【保护服务器内存和 IO】，防止瞬间读取过多群聊记录导致程序崩溃（OOM）。"
+            },
+            "max_concurrent_llm": {
+                "type": "int",
+                "description": "最大 LLM 请求并发数（API 闸口）",
+                "default": 2,
+                "hint": "限制同时发起的 AI 分析请求数量。主要用于【遵守 API 频率限制（RPM）】，避免因为请求过快导致 API 被封号或报错。"
+            },
+            "max_concurrent_t2i": {
+                "type": "int",
+                "description": "最大渲染并发数（物理资源闸口）",
+                "default": 1,
+                "hint": "限制同时开启的浏览器画图进程数。由于渲染图片极其消耗【CPU 和内存】，如果您的服务器只有 1-2G 内存，请务必保持为 1。"
+            }
+        }
    }
 }
@@ -19,6 +19,7 @@ from .src.application.commands.template_command_service import (
 )
 from .src.application.services.analysis_application_service import (
    AnalysisApplicationService,
+    DuplicateGroupTaskError,
 )
 from .src.application.services.message_processing_service import (
    MessageProcessingService,
@@ -515,7 +516,7 @@ class GroupDailyAnalysis(Star):
                if not await adapter.send_text(group_id, text_report):
                    yield event.plain_result(text_report)

-        except asyncio.CancelledError:
+        except DuplicateGroupTaskError:
            yield event.plain_result("📊 该群的分析任务正在执行中，请稍后再试哦~")
        except Exception as e:
            logger.error(f"群分析失败: {e}", exc_info=True)
@@ -789,7 +790,7 @@ class GroupDailyAnalysis(Star):
            try:
                await self.auto_scheduler._perform_auto_analysis_for_group(group_id)
                yield event.plain_result("✅ 自动分析测试完成，请查看群消息")
-            except asyncio.CancelledError:
+            except DuplicateGroupTaskError:
                yield event.plain_result("📊 该群的分析任务正在执行中，请稍后再试哦~")
            except Exception as e:
                yield event.plain_result(f"❌ 自动分析测试失败: {str(e)}")
@@ -24,6 +24,12 @@ from ...infrastructure.persistence.incremental_store import IncrementalStore
 from ...utils.logger import logger


+class DuplicateGroupTaskError(Exception):
+    """当同一个群组在同一时间尝试启动相同类型的重复分析任务时抛出。"""
+
+    pass
+
+
 class AnalysisApplicationService:
    """分析应用服务 - 协调业务流程（每日分析 + 增量分析）"""

@@ -50,7 +56,8 @@ class AnalysisApplicationService:
        self.incremental_merge_service = incremental_merge_service
        self._locks = weakref.WeakValueDictionary()
        # 全局 LLM 分析信号量，控制对外 API 的并发压力
-        max_concurrent = self.config_manager.get_max_concurrent_tasks()
+        # 使用专用的 LLM 并发配置项
+        max_concurrent = self.config_manager.get_llm_max_concurrent()
        self.llm_semaphore = asyncio.Semaphore(max_concurrent)

    @asynccontextmanager
@@ -61,17 +68,18 @@ class AnalysisApplicationService:
        """
        lock_key = f"{task_type}:{group_id}"

-        # 获取或创建该群专属的异步锁
-        lock = self._locks.get(lock_key)
-        if lock is None:
-            lock = asyncio.Lock()
-            self._locks[lock_key] = lock
+        # 获取或创建该群组特有的锁
+        if lock_key not in self._locks:
+            self._locks[lock_key] = asyncio.Lock()
+        lock = self._locks[lock_key]

-        # 检查是否已经锁定（防止并发）
+        # 检查是否已经锁定（防止并发现实）
+        # 说明：在 asyncio 中，虽然是单线程，但设计上应避免阻塞等待非预期的任务。
+        # 这里使用 locked() 检查并立即抛出异常，实现“跳过”而非“排队”。
        if lock.locked():
            logger.warning(f"群 {group_id} 的 {task_type} 任务已在运行，跳过本次请求")
-            # 这里抛出异常以便上层识别并优雅跳过
-            raise asyncio.CancelledError(f"Duplicate task for {lock_key}")
+            # 使用自定义异常以便上层识别并优雅跳过，同时不影响真实的任务取消语义
+            raise DuplicateGroupTaskError(f"Duplicate task for {lock_key}")

        async with lock:
            logger.debug(f"[Lock] 已获取群 {group_id} 的 {task_type} 排他锁")
@@ -317,12 +317,20 @@ class ConfigManager:
        self.config.save_config()

    def get_max_concurrent_tasks(self) -> int:
-        """获取自动分析最大并发数"""
-        return self._get_group("auto_analysis").get("max_concurrent_tasks", 3)
+        """获取自动分析最大并发群数"""
+        return self._get_group("performance").get("max_concurrent_groups", 3)
+
+    def get_llm_max_concurrent(self) -> int:
+        """获取全局 LLM 最大并发请求数"""
+        return self._get_group("performance").get("max_concurrent_llm", 3)
+
+    def get_t2i_max_concurrent(self) -> int:
+        """获取全局图片渲染（T2I）最大并发数"""
+        return self._get_group("performance").get("max_concurrent_t2i", 1)

    def set_max_concurrent_tasks(self, count: int):
        """设置自动分析最大并发数"""
-        self._ensure_group("auto_analysis")["max_concurrent_tasks"] = count
+        self._ensure_group("performance")["max_concurrent_groups"] = count
        self.config.save_config()

    def set_max_messages(self, count: int):
@@ -26,7 +26,8 @@ class ReportGenerator(IReportGenerator):
        self.activity_visualizer = ActivityVisualizer()
        self.html_templates = HTMLTemplates(config_manager)  # 实例化HTML模板管理器
        # 全局 T2I 渲染信号量，保护本地资源
-        max_concurrent = self.config_manager.get_max_concurrent_tasks()
+        # 使用专用的 T2I 并发配置项
+        max_concurrent = self.config_manager.get_t2i_max_concurrent()
        self._render_semaphore = asyncio.Semaphore(max_concurrent)

    async def generate_image_report(
@@ -9,6 +9,7 @@ from typing import Any

 from apscheduler.triggers.cron import CronTrigger

+from ...application.services.analysis_application_service import DuplicateGroupTaskError
 from ...utils.logger import logger
 from ...utils.trace_context import TraceContext
 from ..messaging.message_sender import MessageSender
@@ -316,12 +317,20 @@ class AutoScheduler:
            # 转为列表以便索引
            target_list = list(enabled_targets)

-            logger.info(f"将为 {len(target_list)} 个群聊并发执行分析")
+            max_concurrent = self.config_manager.get_max_concurrent_tasks()
+            sem = asyncio.Semaphore(max_concurrent)
+            logger.info(f"自动分析任务入口并发限制: {max_concurrent}")
+
+            async def throttled_analysis(gid, pid):
+                async with sem:
+                    return await self._perform_auto_analysis_for_group_with_timeout(
+                        gid, pid
+                    )

            analysis_tasks = []
            for gid, pid in target_list:
                task = asyncio.create_task(
-                    self._perform_auto_analysis_for_group_with_timeout(gid, pid),
+                    throttled_analysis(gid, pid),
                    name=f"analysis_group_{gid}",
                )
                analysis_tasks.append(task)
@@ -336,7 +345,7 @@ class AutoScheduler:

            for i, result in enumerate(results):
                gid, _ = target_list[i]
-                if isinstance(result, asyncio.CancelledError):
+                if isinstance(result, DuplicateGroupTaskError):
                    # 锁冲突导致的跳过
                    skip_count += 1
                elif isinstance(result, Exception):
@@ -411,8 +420,8 @@ class AutoScheduler:

            logger.info(f"群 {group_id} 自动分析任务执行成功")

-        except asyncio.CancelledError:
-            # group_lock 抛出的 CancelledError 表示任务正在运行，优雅跳过
+        except DuplicateGroupTaskError:
+            # group_lock 抛出的 DuplicateGroupTaskError 表示任务正在运行，优雅跳过
            logger.debug(f"群 {group_id} 任务因并发锁冲突而跳过（已在运行）")
            raise  # 重新抛出，让上层知道任务并没真正执行而是跳过了
        except Exception as e:
@@ -444,31 +453,33 @@ class AutoScheduler:
                f"(并发限制: {max_concurrent}, 交错间隔: {stagger}秒)"
            )

-            # 资源限制现在由 Application Service 全局控制，此处仅保留交错逻辑
+            # 任务粒度的入口并发限制，保护本地资源（DB/内存）
+            sem = asyncio.Semaphore(max_concurrent)

            async def staggered_incremental(idx, gid, pid):
-                # 按索引交错延迟，均匀分散 API 压力
-                if idx > 0 and stagger > 0:
-                    await asyncio.sleep(stagger * idx)
+                async with sem:
+                    # 按索引交错延迟，均匀分散 API 压力
+                    if idx > 0 and stagger > 0:
+                        await asyncio.sleep(stagger * idx)

-                result = (
-                    await self._perform_incremental_analysis_for_group_with_timeout(
-                        gid, pid
-                    )
-                )
-
-                # 检查是否需要立即发送报告（调试模式）
-                if self.config_manager.get_incremental_report_immediately():
-                    if isinstance(result, dict) and result.get("success"):
-                        logger.info(
-                            f"增量分析立即报告模式生效，正在为群 {gid} 生成报告..."
-                        )
-                        # 立即生成最终报告
-                        await self._perform_incremental_final_report_for_group_with_timeout(
+                    result = (
+                        await self._perform_incremental_analysis_for_group_with_timeout(
                            gid, pid
                        )
+                    )

-                return result
+                    # 检查是否需要立即发送报告（调试模式）
+                    if self.config_manager.get_incremental_report_immediately():
+                        if isinstance(result, dict) and result.get("success"):
+                            logger.info(
+                                f"增量分析立即报告模式生效，正在为群 {gid} 生成报告..."
+                            )
+                            # 立即生成最终报告
+                            await self._perform_incremental_final_report_for_group_with_timeout(
+                                gid, pid
+                            )
+
+                    return result

            analysis_tasks = []
            for idx, (gid, pid) in enumerate(target_list):
@@ -488,7 +499,10 @@ class AutoScheduler:

            for i, result in enumerate(results):
                gid, _ = target_list[i]
-                if isinstance(result, Exception):
+                if isinstance(result, DuplicateGroupTaskError):
+                    # 锁冲突导致的跳过
+                    skip_count += 1
+                elif isinstance(result, Exception):
                    logger.error(f"群 {gid} 增量分析任务异常: {result}")
                    error_count += 1
                elif isinstance(result, dict) and not result.get("success", True):
@@ -563,8 +577,8 @@ class AutoScheduler:
            )
            return result

-        except asyncio.CancelledError:
-            # group_lock 抛出的 CancelledError 表示任务正在运行，优雅跳过
+        except DuplicateGroupTaskError:
+            # group_lock 抛出的 DuplicateGroupTaskError 表示任务正在运行，优雅跳过
            logger.debug(f"群 {group_id} 增量分析因并发锁冲突而跳过（已在运行）")
            return {"success": False, "reason": "already_running"}
        except Exception as e:
@@ -597,14 +611,15 @@ class AutoScheduler:
                f"(并发限制: {max_concurrent}, 交错间隔: {stagger}秒)"
            )

+            sem = asyncio.Semaphore(max_concurrent)
+
            async def staggered_final_report(idx, gid, pid):
-                if idx > 0 and stagger > 0:
-                    await asyncio.sleep(stagger * idx)
-                return (
-                    await self._perform_incremental_final_report_for_group_with_timeout(
+                async with sem:
+                    if idx > 0 and stagger > 0:
+                        await asyncio.sleep(stagger * idx)
+                    return await self._perform_incremental_final_report_for_group_with_timeout(
                        gid, pid
                    )
-                )

            report_tasks = []
            for idx, (gid, pid) in enumerate(target_list):
@@ -624,7 +639,10 @@ class AutoScheduler:

            for i, result in enumerate(results):
                gid, _ = target_list[i]
-                if isinstance(result, Exception):
+                if isinstance(result, DuplicateGroupTaskError):
+                    # 锁冲突导致的跳过
+                    skip_count += 1
+                elif isinstance(result, Exception):
                    logger.error(f"群 {gid} 最终报告任务异常: {result}")
                    error_count += 1
                elif isinstance(result, dict) and not result.get("success", True):
@@ -722,8 +740,8 @@ class AutoScheduler:
            logger.info(f"群 {group_id} 增量最终报告发送成功")
            return result

-        except asyncio.CancelledError:
-            # group_lock 抛出的 CancelledError 表示任务正在运行，优雅跳过
+        except DuplicateGroupTaskError:
+            # group_lock 抛出的 DuplicateGroupTaskError 表示任务正在运行，优雅跳过
            logger.debug(f"群 {group_id} 最终报告因并发锁冲突而跳过（已在运行）")
            return {"success": False, "reason": "already_running"}
        except Exception as e: