diff --git a/flows/shared/clients.py b/flows/shared/clients.py index b414e5e..ab8d673 100644 --- a/flows/shared/clients.py +++ b/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()