From 7850697f1ae4892524ef5cb3e5476c002b2fa5a1 Mon Sep 17 00:00:00 2001 From: SXP-Simon Date: Tue, 3 Mar 2026 20:51:47 +0800 Subject: [PATCH] =?UTF-8?q?refactor(perf):=20=E5=BB=BA=E7=AB=8B=E4=B8=89?= =?UTF-8?q?=E7=BA=A7=E8=B5=84=E6=BA=90=E9=98=B2=E7=BA=BF=E5=B9=B6=E4=BC=98?= =?UTF-8?q?=E5=8C=96=E5=B9=B6=E5=8F=91=E9=99=90=E6=B5=81=E9=85=8D=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 【异常处理优化】引入自定义任务重复异常，精准区分“因重复运行而跳过”与“实际分析取消”两种不同语义，提升调度健壮性。 - 【三级并发控制架构】针对不同层级的资源消耗，建立了完整的限流保护体系： - 入口层：控制同时处于活跃状态的任务总数，防止瞬时加载大量消息导致服务器内存溢出 (OOM)； - 接口层：全局限制 LLM API 并发请求数，确保严格遵守服务商的频率限制 (RPM)； - 渲染层：全局限制浏览器图片生成进程，保护物理 CPU 与内存不被高能耗渲染任务耗尽。 - 【配置体验升级】新增「并发限流设置」配置组，统一收纳核心性能参数，并提供面向普通用户的防坑说明与更稳健的默认值。 - 【代码清理】移除陈旧的兼容逻辑与冗余参数，统一配置命名规范。 - 【功能精化】重排增量分析逻辑，确保在任务入口与分段处理流程中均有可靠的信号量保护。 - 【细节完善】优化群文件及群相册上传设置的文案描述，明确功能支持范围与权限要求。 --- _conf_schema.json | 31 +++++-- main.py | 5 +- .../services/analysis_application_service.py | 26 ++++-- src/infrastructure/config/config_manager.py | 14 ++- src/infrastructure/reporting/generators.py | 3 +- .../scheduler/auto_scheduler.py | 88 +++++++++++-------- 6 files changed, 111 insertions(+), 56 deletions(-) diff --git a/_conf_schema.json b/_conf_schema.json index c1132f4..579de05 100644 --- a/_conf_schema.json +++ b/_conf_schema.json @@ -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。" + } + } } } diff --git a/main.py b/main.py index cd846ed..0105748 100644 --- a/main.py +++ b/main.py @@ -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)}") diff --git a/src/application/services/analysis_application_service.py b/src/application/services/analysis_application_service.py index 54fe2bc..c5196af 100644 --- a/src/application/services/analysis_application_service.py +++ b/src/application/services/analysis_application_service.py @@ -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} 排他锁") diff --git a/src/infrastructure/config/config_manager.py b/src/infrastructure/config/config_manager.py index 281a9b6..74c0b37 100644 --- a/src/infrastructure/config/config_manager.py +++ b/src/infrastructure/config/config_manager.py @@ -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): diff --git a/src/infrastructure/reporting/generators.py b/src/infrastructure/reporting/generators.py index 1ada24f..a908b27 100644 --- a/src/infrastructure/reporting/generators.py +++ b/src/infrastructure/reporting/generators.py @@ -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( diff --git a/src/infrastructure/scheduler/auto_scheduler.py b/src/infrastructure/scheduler/auto_scheduler.py index 6646547..a7063c2 100644 --- a/src/infrastructure/scheduler/auto_scheduler.py +++ b/src/infrastructure/scheduler/auto_scheduler.py @@ -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: