diff --git a/flows/shared/clients.py b/flows/shared/clients.py index 74f9dc3..ab8d673 100644 --- a/flows/shared/clients.py +++ b/flows/shared/clients.py @@ -8,6 +8,8 @@ from __future__ import annotations import json +import random +import threading import time from datetime import datetime, timedelta, timezone from typing import Any, Callable, TypeVar @@ -21,7 +23,9 @@ from flows.shared.mappers import map_sds_channel_type T = TypeVar("T") -GOODS_CONCURRENCY = 20 +GOODS_CONCURRENCY = 5 +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 @@ -144,8 +148,25 @@ def fetch_sds_all_pages(client: SdsMngClient, params: dict[str, Any]) -> list[An # ---------- RIIN 商品 ---------- +# 全局限速:RIIN 分销侧反馈 distribution/order/goods 逐单调用对其服务器压力过大, +# 所有线程共享节拍,相邻两次调用间隔 = GOODS_MIN_INTERVAL_MS + 随机抖动(340~500ms,峰值 < 3 次/秒)。 +_goods_pace_lock = threading.Lock() +_goods_last_call_at = 0.0 + + +def _pace_goods_call() -> None: + global _goods_last_call_at + with _goods_pace_lock: + 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() + + def fetch_goods_for_order(client: RiinDisClient, order_id: str) -> list[Any]: try: + _pace_goods_call() resp = client.get_order_goods(order_id) return list(resp.data) except Exception as err: # noqa: BLE001 diff --git a/prefect.yaml b/prefect.yaml index 1d1c04a..b1f2353 100644 --- a/prefect.yaml +++ b/prefect.yaml @@ -21,6 +21,7 @@ deployments: schedules: - cron: 30 * * * * timezone: Asia/Shanghai + active: false # 暂停:每小时重拉近7天全部订单商品,对 RIIN distribution/order/goods 调用量过大(增量/刷新未完结/日报核验已覆盖) work_pool: name: inkreach-pool - name: riin-refresh-pending @@ -39,6 +40,7 @@ deployments: schedules: - cron: 0 6 * * * timezone: Asia/Shanghai + active: false # 暂停:与 daily-report 18:30 的核验子流程重复,保留日报一处即可 work_pool: name: inkreach-pool - name: sds-order-sync