feat: ocg_router 并发模式——X-OCG-Mode: concurrent 时 round-robin 锁 key 轮询(每次请求轮着用key,避免并发压同一key触发429/401), 非并发模式保持按用量选最空闲; 测试验证key1→key4→key5→key6轮询
This commit is contained in:
@@ -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,10 +454,37 @@ 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)
|
||||
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"])
|
||||
|
||||
Reference in New Issue
Block a user