fix: 修复QPS调度器死锁问题，优化请求获取流程

问题：当就绪队列达到max_prefetch上限时，所有工作线程都卡在
submit()的背压循环中，无法调用get_ready_request()消费队列，
导致死锁。

解决方案：
1. 调整_get_request流程：先尝试从就绪队列获取，再提交新请求
2. submit使用非阻塞模式(block=False)，避免背压死锁
3. 新增调度器状态日志，便于监控和调试

Author: ShellMonster

feapder/core/parser_control.py

@@ -72,28 +72,41 @@ def _get_request(self):
         """
         @summary: 获取请求
                   如果启用了QPS限流，流程为：
-                  1. 从Collector批量取出请求
-                  2. 提交到QPSScheduler进行限流调度
-                  3. 从QPSScheduler的就绪队列获取已获得令牌的请求
+                  1. 优先从就绪队列获取请求（避免死锁）
+                  2. 就绪队列为空时，从Collector取请求提交给调度器
+                  3. 再次尝试从就绪队列获取请求
                   如果未启用QPS限流，直接从Collector获取
         ---------
         @result: 请求字典，包含request_obj和request_redis，队列为空返回None
         """
         if self._qps_scheduler:
             # QPS限流模式
-            # 步骤1：从Collector获取请求并提交给调度器
+            # 步骤1：优先从就绪队列获取请求（非阻塞，避免死锁）
+            request = self._qps_scheduler.get_ready_request_nowait()
+            if request:
+                return request
+
+            # 步骤2：就绪队列为空，从Collector获取请求并提交给调度器
             # Collector.get_request() 返回的是 dict: {"request_obj": Request, "request_redis": str}
             batch_size = max(1, self._qps_scheduler.max_prefetch // 10)
             for _ in range(batch_size):
                 request_dict = self._collector.get_request()
                 if request_dict:
-                    # 将请求字典包装后提交给调度器
-                    self._qps_scheduler.submit(request_dict)
+                    # 非阻塞提交，避免背压导致死锁
+                    self._qps_scheduler.submit(request_dict, block=False)
                 else:
                     break
 
-            # 步骤2：从调度器获取已就绪的请求（与原始模式timeout保持一致）
-            return self._qps_scheduler.get_ready_request(timeout=1.0)
+            # 步骤3：从调度器获取已就绪的请求（带超时）
+            request = self._qps_scheduler.get_ready_request(timeout=1.0)
+
+            # 超时返回None时输出调试日志
+            if request is None and not self._qps_scheduler.is_empty():
+                log.debug(
+                    f"QPS: 等待就绪请求超时，调度器中仍有 {self._qps_scheduler.pending_count()} 个请求在等待令牌"
+                )
+
+            return request
         else:
             # 原始模式：直接从Collector获取
             return self._collector.get_request()
@@ -526,27 +539,41 @@ def _get_request(self):
         """
         @summary: 获取请求
                   如果启用了QPS限流，流程为：
-                  1. 从MemoryDB批量取出请求
-                  2. 提交到QPSScheduler进行限流调度
-                  3. 从QPSScheduler的就绪队列获取已获得令牌的请求
+                  1. 优先从就绪队列获取请求（避免死锁）
+                  2. 就绪队列为空时，从MemoryDB取请求提交给调度器
+                  3. 再次尝试从就绪队列获取请求
                   如果未启用QPS限流，直接从MemoryDB获取
         ---------
         @result: 请求对象Request，队列为空返回None
         """
         if self._qps_scheduler:
             # QPS限流模式
-            # 步骤1：从MemoryDB批量获取请求并提交给调度器
+            # 步骤1：优先从就绪队列获取请求（非阻塞，避免死锁）
+            request = self._qps_scheduler.get_ready_request_nowait()
+            if request:
+                return request
+
+            # 步骤2：就绪队列为空，从MemoryDB获取请求并提交给调度器
             # 批量大小与max_prefetch相关，避免过多请求堆积在调度器中
             batch_size = max(1, self._qps_scheduler.max_prefetch // 10)
             for _ in range(batch_size):
                 pending_request = self._memory_db.get_nowait()
                 if pending_request:
-                    self._qps_scheduler.submit(pending_request)
+                    # 非阻塞提交，避免背压导致死锁
+                    self._qps_scheduler.submit(pending_request, block=False)
                 else:
                     break
 
-            # 步骤2：从调度器获取已就绪的请求（与原始模式timeout保持一致）
-            return self._qps_scheduler.get_ready_request(timeout=1.0)
+            # 步骤3：从调度器获取已就绪的请求（带超时）
+            request = self._qps_scheduler.get_ready_request(timeout=1.0)
+
+            # 超时返回None时输出调试日志
+            if request is None and not self._qps_scheduler.is_empty():
+                log.debug(
+                    f"QPS: 等待就绪请求超时，调度器中仍有 {self._qps_scheduler.pending_count()} 个请求在等待令牌"
+                )
+
+            return request
         else:
             # 原始模式：直接从MemoryDB获取
             return self._memory_db.get()

feapder/core/schedulers/qps_scheduler.py

@@ -95,6 +95,10 @@ def __init__(
             'ready': 0,           # 已就绪的请求数
         }
 
+        # 状态日志控制
+        self._last_status_log_time = 0
+        self._status_log_interval = 10  # 每10秒输出一次状态
+
     def start(self) -> None:
         """
         @summary: 启动调度器，启动后台调度线程开始处理请求的QPS限流
@@ -150,6 +154,7 @@ def submit(self, request: Any, block: bool = True, timeout: float = None) -> boo
         # 检查是否需要背压
         if self.max_prefetch > 0:
             start_time = time.time()
+            logged_backpressure = False
             while True:
                 with self._count_lock:
                     if self._domain_pending_count[domain] < self.max_prefetch:
@@ -162,6 +167,13 @@ def submit(self, request: Any, block: bool = True, timeout: float = None) -> boo
                 if timeout is not None and (time.time() - start_time) >= timeout:
                     return False
 
+                # 背压阻塞日志（只输出一次，避免刷屏）
+                if not logged_backpressure:
+                    log.debug(
+                        f"QPS背压: 域名 {domain} 待处理数达到上限 {self.max_prefetch}，等待释放..."
+                    )
+                    logged_backpressure = True
+
                 # 等待一小段时间后重试
                 time.sleep(0.01)
         else:
@@ -223,6 +235,9 @@ def _scheduler_loop(self) -> None:
                 # 处理延迟堆（将到期的请求移入就绪队列）
                 moved = self._process_delay_heap()
 
+                # 定期输出状态日志（DEBUG模式下）
+                self._log_status_if_needed()
+
                 # 如果没有任何工作，短暂休眠避免CPU空转
                 if not processed and not moved:
                     time.sleep(0.001)  # 1ms
@@ -290,6 +305,30 @@ def _process_delay_heap(self) -> int:
 
         return count
 
+    def _log_status_if_needed(self) -> None:
+        """
+        @summary: 定期输出调度器状态日志（DEBUG模式下）
+        ---------
+        @result:
+        """
+        now = time.time()
+        if now - self._last_status_log_time >= self._status_log_interval:
+            self._last_status_log_time = now
+
+            with self._heap_lock:
+                delay_heap_size = len(self._delay_heap)
+
+            submit_queue_size = self._submit_queue.qsize()
+            ready_queue_size = self._ready_queue.qsize()
+
+            # 只有当有请求在处理时才输出日志
+            if delay_heap_size > 0 or submit_queue_size > 0 or ready_queue_size > 0:
+                log.debug(
+                    f"QPS调度器状态: 提交队列={submit_queue_size}, "
+                    f"延迟堆={delay_heap_size}, 就绪队列={ready_queue_size}, "
+                    f"统计={self._stats}"
+                )
+
     def is_empty(self) -> bool:
         """
         @summary: 检查调度器是否为空