flows/shared/clients.py

@@ -8,6 +8,7 @@
 from __future__ import annotations
 
 import json
+import random
 import threading
 import time
 from datetime import datetime, timedelta, timezone
@@ -23,7 +24,8 @@ from flows.shared.mappers import map_sds_channel_type
 T = TypeVar("T")
 
 GOODS_CONCURRENCY = 5
-GOODS_MIN_INTERVAL_MS = 150  # 相邻两次 order/goods 调用全局最小间隔(峰值 ≤ ~6.7 次/秒)
+GOODS_MIN_INTERVAL_MS = 340  # 相邻两次 order/goods 调用全局最小间隔(峰值 < 3 次/秒)
+GOODS_MAX_JITTER_MS = 160  # 在最小间隔上叠加的随机抖动上限,实际间隔 340~500ms 随机
 PAGE_DELAY_MS = 500
 MAX_RETRIES = 3
 RETRY_DELAY_MS = 3000
@@ -147,7 +149,7 @@ def fetch_sds_all_pages(client: SdsMngClient, params: dict[str, Any]) -> list[An
 # ---------- RIIN 商品 ----------
 
 # 全局限速:RIIN 分销侧反馈 distribution/order/goods 逐单调用对其服务器压力过大,
-# 所有线程共享节拍,相邻两次调用至少间隔 GOODS_MIN_INTERVAL_MS。
+# 所有线程共享节拍,相邻两次调用间隔 = GOODS_MIN_INTERVAL_MS + 随机抖动(340~500ms,峰值 < 3 次/秒)。
 _goods_pace_lock = threading.Lock()
 _goods_last_call_at = 0.0
 
@@ -155,7 +157,8 @@ _goods_last_call_at = 0.0
 def _pace_goods_call() -> None:
     global _goods_last_call_at
     with _goods_pace_lock:
-        wait = _goods_last_call_at + GOODS_MIN_INTERVAL_MS / 1000 - time.monotonic()
+        interval = (GOODS_MIN_INTERVAL_MS + random.uniform(0, GOODS_MAX_JITTER_MS)) / 1000
+        wait = _goods_last_call_at + interval - time.monotonic()
         if wait > 0:
             time.sleep(wait)
         _goods_last_call_at = time.monotonic()