From 9b26a3e5c9f1c9ad67d9c81407fc3c557095fe49 Mon Sep 17 00:00:00 2001 From: SXP-Simon Date: Tue, 3 Mar 2026 20:56:24 +0800 Subject: [PATCH] =?UTF-8?q?fix(perf):=20=E5=BC=95=E5=85=A5=E5=8E=9F?= =?UTF-8?q?=E5=AD=90=E5=8C=96=E4=BB=BB=E5=8A=A1=E6=B3=A8=E5=86=8C=E8=A1=A8?= =?UTF-8?q?=EF=BC=8C=E6=B6=88=E9=99=A4=E5=BC=82=E6=AD=A5=E9=94=81=E7=8A=B6?= =?UTF-8?q?=E6=80=81=E6=A3=80=E6=9F=A5=E7=9A=84=E7=AB=9E=E6=80=81=E7=9B=B2?= =?UTF-8?q?=E5=8C=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 【深度优化】将原有的 'if lock.locked()' 探测逻辑重构为基于同步集合的「任务注册表」模式。 - 【原理解修】规避了 asyncio 环境下异步锁获取的非原子性问题： - 痛点：虽然 asyncio 是单线程，但 `async with lock` 内部的 `acquire()` 是一个 `await` 点； - 风险：在高并发瞬间，多个协程可能在执行 `await lock.acquire()` 时让出控制权，导致它们都通过了之前的 `locked()` 检查，最终在锁上排队而非直接跳过。 - 【原子性方案】利用 asyncio 事件循环中“同步代码块不可被中断”的特性，将任务的「检查」与「占位」合并为连续的同步操作。 - 【确定性控制】确保重复请求在触达任何 `await` 点之前即被同步拦截，从而 100% 保证并发冲突时触发“直接跳过”的语义，彻底杜绝意外排队现象。 --- .../services/analysis_application_service.py | 27 +++++++++++-------- 1 file changed, 16 insertions(+), 11 deletions(-) diff --git a/src/application/services/analysis_application_service.py b/src/application/services/analysis_application_service.py index c5196af..66a4cb6 100644 --- a/src/application/services/analysis_application_service.py +++ b/src/application/services/analysis_application_service.py @@ -59,6 +59,8 @@ class AnalysisApplicationService: # 使用专用的 LLM 并发配置项 max_concurrent = self.config_manager.get_llm_max_concurrent() self.llm_semaphore = asyncio.Semaphore(max_concurrent) + # 用于追踪当前正在执行的任务，实现原子的“检查并设置”逻辑，避免 locked() 竞态 + self._active_tasks = set() @asynccontextmanager async def group_lock(self, group_id: str, task_type: str = "analysis"): @@ -68,25 +70,28 @@ class AnalysisApplicationService: """ lock_key = f"{task_type}:{group_id}" - # 获取或创建该群组特有的锁 + # 获取或创建该群组特有的锁（保留锁作为第二道资源限流防线） if lock_key not in self._locks: self._locks[lock_key] = asyncio.Lock() lock = self._locks[lock_key] - # 检查是否已经锁定（防止并发现实） - # 说明：在 asyncio 中，虽然是单线程，但设计上应避免阻塞等待非预期的任务。 - # 这里使用 locked() 检查并立即抛出异常，实现“跳过”而非“排队”。 - if lock.locked(): + # 使用同步集合实现原子化的“运行中”检查 + # 在 asyncio 的单线程循环中，同步代码段不会被中断，因此这是原子操作 + if lock_key in self._active_tasks: logger.warning(f"群 {group_id} 的 {task_type} 任务已在运行，跳过本次请求") - # 使用自定义异常以便上层识别并优雅跳过，同时不影响真实的任务取消语义 raise DuplicateGroupTaskError(f"Duplicate task for {lock_key}") - async with lock: - logger.debug(f"[Lock] 已获取群 {group_id} 的 {task_type} 排他锁") - try: + # 占位：标记任务开始 + self._active_tasks.add(lock_key) + + try: + async with lock: + logger.debug(f"[Lock] 已获取群 {group_id} 的 {task_type} 排他锁") yield - finally: - logger.debug(f"[Lock] 已释放群 {group_id} 的 {task_type} 排他锁") + finally: + # 释放：标记任务结束 + self._active_tasks.discard(lock_key) + logger.debug(f"[Lock] 已释放群 {group_id} 的 {task_type} 排他锁") async def execute_daily_analysis( self, group_id: str, platform_id: str | None = None, manual: bool = False