方案修改为 多线程 滚动

app/services/job_clean_service.py

@@ -1,4 +1,8 @@
-"""岗位清洗服务（协程版）"""
+"""岗位清洗服务（持续消费模型）
+
+启动 N 个 worker 协程，每个 worker 循环：取一条 → 处理 → 取下一条。
+没数据时短暂休眠后重试。
+"""
 
 import asyncio
 import json
@@ -28,70 +32,83 @@ _company_lock = asyncio.Lock()
 # 累计已清洗总数（用于总数上限控制，进程内统计）
 _cleaned_total = 0
 
+# 停止信号
+_stop_event = asyncio.Event()
+
 
 def is_clean_limit_reached() -> bool:
     """是否已达到累计清洗总数上限（clean_total_limit=0 表示不限制）"""
     return 0 < settings.clean_total_limit <= _cleaned_total
 
 
+def stop_job_clean():
+    """外部调用，通知所有 worker 停止"""
+    _stop_event.set()
+
+
 async def run_job_clean() -> None:
-    """一次批量清洗任务"""
+    """启动 N 个 worker 协程持续消费，直到收到停止信号或达到上限"""
+    _stop_event.clear()
+    worker_count = settings.clean_worker_count
+    log.info("岗位清洗：启动 {} 个 worker 协程", worker_count)
+
+    workers = [asyncio.create_task(_worker(i)) for i in range(worker_count)]
+    await asyncio.gather(*workers)
+
+    log.info("岗位清洗：所有 worker 已退出，累计清洗 {} 条", _cleaned_total)
+
+
+async def _worker(worker_id: int) -> None:
+    """单个 worker：循环取一条、处理一条"""
     global _cleaned_total
 
-    # 总数上限：已达上限直接跳过
-    if is_clean_limit_reached():
-        return
+    while not _stop_event.is_set():
+        # 总数上限检查
+        if is_clean_limit_reached():
+            return
 
-    # 1. 从 PG 锁定一批待清洗数据
+        # 从 PG 取一条待处理数据
+        data = await _fetch_one()
+        if data is None:
+            # 没数据，休眠后重试
+            await asyncio.sleep(settings.clean_idle_sleep)
+            continue
+
+        # 处理
+        try:
+            await _do_clean(data)
+            _cleaned_total += 1
+        except Exception as e:
+            log.error("[worker-{}] 岗位清洗异常, id={}: {}", worker_id, data["id"], e)
+
+        # 达到上限后通知所有 worker 停止
+        if is_clean_limit_reached():
+            log.info("岗位清洗：已达累计上限 {} 条，停止", settings.clean_total_limit)
+            _stop_event.set()
+            return
+
+
+async def _fetch_one() -> dict | None:
+    """从 PG 锁定一条待清洗数据并标记为 cleaning"""
     async with PgSession() as pg:
         result = await pg.execute(
-            text("""
-                SELECT * FROM app_job_data
-                WHERE clean_status = 'pending' AND  recruit_category in (0,1,2)
-                LIMIT :limit
-                FOR UPDATE SKIP LOCKED
-            """),
-            {"limit": settings.clean_batch_size},
-        )
-        rows = result.mappings().all()
-        if not rows:
-            return
-
-        ids = [r["id"] for r in rows]
-        await pg.execute(
             text("""
                 UPDATE app_job_data
                 SET clean_status = 'cleaning', clean_started_at = NOW()
-                WHERE id = ANY(:ids)
+                WHERE id = (
+                    SELECT id FROM app_job_data
+                    WHERE clean_status = 'pending' AND recruit_category IN (0,1,2)
+                    LIMIT 1
+                    FOR UPDATE SKIP LOCKED
+                )
+                RETURNING *
             """),
-            {"ids": ids},
         )
-        await pg.commit()
-
-    log.info("岗位清洗：锁定{}条数据", len(rows))
-
-    # 2. 协程并发清洗，信号量限流
-    sem = asyncio.Semaphore(settings.clean_concurrency)
-    tasks = [_clean_one(sem, dict(r)) for r in rows]
-    results = await asyncio.gather(*tasks, return_exceptions=True)
-
-    # 汇总
-    errors = sum(1 for r in results if isinstance(r, Exception))
-    _cleaned_total += len(rows)
-    log.info("岗位清洗：本批完成，共{}条，异常{}条，累计{}条", len(rows), errors, _cleaned_total)
-
-    if is_clean_limit_reached():
-        log.info("岗位清洗：已达累计上限 {} 条，停止清洗任务", settings.clean_total_limit)
-
-
-async def _clean_one(sem: asyncio.Semaphore, data: dict) -> None:
-    """单条岗位清洗"""
-    async with sem:
-        try:
-            await _do_clean(data)
-        except Exception as e:
-            log.error("岗位清洗异常, id={}: {}", data["id"], e)
-            # 保持 cleaning 状态，由僵尸恢复任务重置
+        row = result.mappings().first()
+        if row:
+            await pg.commit()
+            return dict(row)
+        return None
 
 
 async def _do_clean(data: dict) -> None:
@@ -253,7 +270,6 @@ async def _extract_skill_tags(job_id: int, result: dict) -> None:
         if not name or len(name) > 50:
             continue
 
-        # 每个 tag 单独 session，避免死锁
         real_id = await _find_or_create_skill_tag(name)
         if real_id and real_id not in tag_ids:
             tag_ids.append(real_id)
@@ -311,7 +327,7 @@ async def _find_or_create_company(short_name: str, urllistid: int | None = None)
             )
             await mysql.commit()
 
-    # 锁外处理 logo：仅新建公司时执行，上传是网络IO，不阻塞其他协程；失败不影响主流程
+    # 锁外处理 logo
     try:
         logo_b64 = await _get_logo_base64(urllistid)
         if logo_b64: