From 6a5c08a5df66dd8909cd18f847454dcd873c042e Mon Sep 17 00:00:00 2001 From: mohe Date: Fri, 14 Aug 2026 04:06:26 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20ocg=5Frouter=20=E5=B9=B6=E5=8F=91?= =?UTF-8?q?=E6=A8=A1=E5=BC=8F=E2=80=94=E2=80=94X-OCG-Mode:=20concurrent=20?= =?UTF-8?q?=E6=97=B6=20round-robin=20=E9=94=81=20key=20=E8=BD=AE=E8=AF=A2(?= =?UTF-8?q?=E6=AF=8F=E6=AC=A1=E8=AF=B7=E6=B1=82=E8=BD=AE=E7=9D=80=E7=94=A8?= =?UTF-8?q?key,=E9=81=BF=E5=85=8D=E5=B9=B6=E5=8F=91=E5=8E=8B=E5=90=8C?= =?UTF-8?q?=E4=B8=80key=E8=A7=A6=E5=8F=91429/401),=20=E9=9D=9E=E5=B9=B6?= =?UTF-8?q?=E5=8F=91=E6=A8=A1=E5=BC=8F=E4=BF=9D=E6=8C=81=E6=8C=89=E7=94=A8?= =?UTF-8?q?=E9=87=8F=E9=80=89=E6=9C=80=E7=A9=BA=E9=97=B2;=20=E6=B5=8B?= =?UTF-8?q?=E8=AF=95=E9=AA=8C=E8=AF=81key1=E2=86=92key4=E2=86=92key5?= =?UTF-8?q?=E2=86=92key6=E8=BD=AE=E8=AF=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- gateway/scripts/ocg_router.py | 66 ++++++++++++++++++++++++++++++++--- 1 file changed, 62 insertions(+), 4 deletions(-) diff --git a/gateway/scripts/ocg_router.py b/gateway/scripts/ocg_router.py index f95a3bf..c8680c4 100644 --- a/gateway/scripts/ocg_router.py +++ b/gateway/scripts/ocg_router.py @@ -70,6 +70,8 @@ _key_failure_reason: dict[str, str] = {} # key_id → 最近失败原因（冷 _route_stats: dict[str, int] = {} # key_id → hit_count _total_hits = 0 _current_key_id = "" +# 2026-08-13 并发模式：round-robin 轮询索引（锁 key 轮着来） +_rr_index = 0 _state_lock = threading.Lock() @@ -202,6 +204,35 @@ def pick_key(): return best +def pick_key_round_robin(): + """并发模式：round-robin 轮询选 key（锁 key 轮着来）。 + 每次请求用下一个 key，避免并发压同一 key（2026-08-13 老莫设计）。 + 跳过不可用 key（失败冷却中/未订阅）。""" + global _rr_index + if not _keys: + return None + with _state_lock: + n = len(_keys) + if n == 0: + return None + # 轮询：从 _rr_index 开始找下一个可用 key + for offset in range(n): + idx = (_rr_index + offset) % n + k = _keys[idx] + kid = k["key_id"] + # 跳过不可用（失败冷却中/未订阅） + if time.time() < _key_failures.get(kid, 0): + continue + usage = _key_usage.get(kid, {}) + if not usage.get("subscribed", True): + continue + # 选中：更新索引到下一个 + _rr_index = (idx + 1) % n + return k + # 全部不可用 → 返回第一个（兜底） + return _keys[0] + + def mark_key_failed(kid, reason="", cooldown=FAIL_COOLDOWN_SEC): """标记 key 故障，进入冷却期。记录失败原因供冷却期报错。""" until = time.time() + cooldown @@ -423,14 +454,41 @@ class RouterHandler(BaseHTTPRequestHandler): except (json.JSONDecodeError, UnicodeDecodeError) as e: log.warning("failed to parse request body (len=%d): %s", len(body), e) + # 2026-08-13 并发模式：header X-OCG-Mode: concurrent → round-robin 锁 key 轮询 + concurrent_mode = self.headers.get("X-OCG-Mode", "").lower() == "concurrent" + # 尝试所有 key（按用量排序），直到成功 with _state_lock: usage_snapshot = dict(_key_usage) failures_snapshot = dict(_key_failures) - ranked_keys = sorted( - list(_keys), - key=lambda k: _key_health_score(k["key_id"]) - ) + if concurrent_mode: + # 并发模式：round-robin 轮询（锁 key 轮着来） + ranked_keys = [] + n = len(_keys) + global _rr_index + for offset in range(n): + idx = (_rr_index + offset) % n + k = _keys[idx] + kid = k["key_id"] + if time.time() < failures_snapshot.get(kid, 0): + continue + usage = usage_snapshot.get(kid, {}) + if not usage.get("subscribed", True): + continue + ranked_keys.append(k) + # 更新轮询索引 + if ranked_keys: + first_kid = ranked_keys[0]["key_id"] + for i, k in enumerate(_keys): + if k["key_id"] == first_kid: + _rr_index = (i + 1) % len(_keys) + break + log.info("concurrent mode: round-robin, next_idx=%d, pool=%d", _rr_index, len(ranked_keys)) + else: + ranked_keys = sorted( + list(_keys), + key=lambda k: _key_health_score(k["key_id"]) + ) last_error = "" success = False