github/claude-relay-service
1
0
Fork 0
mirror of https://github.com/Wei-Shaw/claude-relay-service.git synced 2026-01-23 09:38:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat: 为 Claude Console 账户添加并发控制机制

Browse Source 
实现了完整的 Claude Console 账户并发任务数控制功能，防止单账户过载，提升服务稳定性。

  **核心功能**

  - 🔒 **原子性并发控制**: 基于 Redis Sorted Set 实现的抢占式并发槽位管理，防止竞态条件
  - 🔄 **自动租约刷新**: 流式请求每 5 分钟自动刷新租约，防止长连接租约过期
  - 🚨 **智能降级处理**: 并发满额时自动清理粘性会话并重试其他账户（最多 1 次）
  - 🎯 **专用错误码**: 引入 `CONSOLE_ACCOUNT_CONCURRENCY_FULL` 错误码，区分并发限制和其他错误
  - 📊 **批量性能优化**: 调度器使用 Promise.all 并行查询账户并发数，减少 Redis 往返

  **后端实现**

  1. **Redis 并发控制方法** (src/models/redis.js)
     - `incrConsoleAccountConcurrency()`: 增加并发计数（带租约）
     - `decrConsoleAccountConcurrency()`: 释放并发槽位
     - `refreshConsoleAccountConcurrencyLease()`: 刷新租约（流式请求）
     - `getConsoleAccountConcurrency()`: 查询当前并发数

  2. **账户服务增强** (src/services/claudeConsoleAccountService.js)
     - 添加 `maxConcurrentTasks` 字段（默认 0 表示无限制）
     - 获取账户时自动查询实时并发数 (`activeTaskCount`)
     - 支持更新并发限制配置

  3. **转发服务并发保护** (src/services/claudeConsoleRelayService.js)
     - 请求前原子性抢占槽位，超限则立即回滚并抛出专用错误
     - 流式请求启动定时器每 5 分钟刷新租约
     - `finally` 块确保槽位释放（即使发生异常）
     - 为每个请求分配唯一 `requestId` 用于并发追踪

  4. **统一调度器优化** (src/services/unifiedClaudeScheduler.js)
     - 获取可用账户时批量查询并发数（Promise.all 并行）
     - 预检查并发限制，避免选择已满的账户
     - 检查分组成员时也验证并发状态
     - 所有账户并发满额时抛出专用错误码

  5. **API 路由降级处理** (src/routes/api.js)
     - 捕获 `CONSOLE_ACCOUNT_CONCURRENCY_FULL` 错误
     - 自动清理粘性会话映射并重试（最多 1 次）
     - 重试失败返回 503 错误和友好提示
     - count_tokens 端点也支持并发满额重试

  6. **管理端点验证** (src/routes/admin.js)
     - 创建/更新账户时验证 `maxConcurrentTasks` 为非负整数
     - 支持前端传入并发限制配置

  **前端实现**

  1. **表单字段** (web/admin-spa/src/components/accounts/AccountForm.vue)
     - 添加"最大并发任务数"输入框（创建和编辑模式）
     - 支持占位符提示"0 表示不限制"
     - 表单数据自动映射到后端 API

  2. **实时监控** (web/admin-spa/src/views/AccountsView.vue)
     - 账户列表显示并发状态进度条和百分比
     - 颜色编码：绿色（<80%）、黄色（80%-100%）、红色（100%）
     - 显示"X / Y"格式的并发数（如"2 / 5"）
     - 未配置限制时显示"并发无限制"徽章
This commit is contained in:
sususu98
2025-10-21 13:43:57 +08:00
parent b61a3103e9
commit 1458d609ca
8 changed files with 706 additions and 119 deletions
Download Patch File Download Diff File Expand all files Collapse all files

127
src/services/claudeConsoleRelayService.js
View File

@@ -1,5 +1,7 @@
const axios = require('axios')
const { v4: uuidv4 } = require('uuid')
const claudeConsoleAccountService = require('./claudeConsoleAccountService')
const redis = require('../models/redis')
const logger = require('../utils/logger')
const config = require('../../config/config')
const {
@@ -25,6 +27,8 @@ class ClaudeConsoleRelayService {
) {
let abortController = null
let account = null
const requestId = uuidv4() // 用于并发追踪
let concurrencyAcquired = false
try {
// 获取账户信息
@@ -34,8 +38,37 @@ class ClaudeConsoleRelayService {
}
logger.info(
`📤 Processing Claude Console API request for key: ${apiKeyData.name || apiKeyData.id}, account: ${account.name} (${accountId})`
`📤 Processing Claude Console API request for key: ${apiKeyData.name || apiKeyData.id}, account: ${account.name} (${accountId}), request: ${requestId}`
)
// 🔒 并发控制：原子性抢占槽位
if (account.maxConcurrentTasks > 0) {
// 先抢占，再检查 - 避免竞态条件
const newConcurrency = Number(
await redis.incrConsoleAccountConcurrency(accountId, requestId, 600)
)
concurrencyAcquired = true
// 检查是否超过限制
if (newConcurrency > account.maxConcurrentTasks) {
// 超限，立即回滚
await redis.decrConsoleAccountConcurrency(accountId, requestId)
concurrencyAcquired = false
logger.warn(
`⚠️ Console account ${account.name} (${accountId}) concurrency limit exceeded: ${newConcurrency}/${account.maxConcurrentTasks} (request: ${requestId}, rolled back)`
)
const error = new Error('Console account concurrency limit reached')
error.code = 'CONSOLE_ACCOUNT_CONCURRENCY_FULL'
error.accountId = accountId
throw error
}
logger.debug(
`🔓 Acquired concurrency slot for account ${account.name} (${accountId}), current: ${newConcurrency}/${account.maxConcurrentTasks}, request: ${requestId}`
)
}
logger.debug(`🌐 Account API URL: ${account.apiUrl}`)
logger.debug(`🔍 Account supportedModels: ${JSON.stringify(account.supportedModels)}`)
logger.debug(`🔑 Account has apiKey: ${!!account.apiKey}`)
@@ -297,6 +330,21 @@ class ClaudeConsoleRelayService {
// 不再因为模型不支持而block账号
throw error
} finally {
// 🔓 并发控制：释放并发槽位
if (concurrencyAcquired) {
try {
await redis.decrConsoleAccountConcurrency(accountId, requestId)
logger.debug(
`🔓 Released concurrency slot for account ${account?.name || accountId}, request: ${requestId}`
)
} catch (releaseError) {
logger.error(
`❌ Failed to release concurrency slot for account ${accountId}, request: ${requestId}:`,
releaseError.message
)
}
}
}
}
@@ -312,6 +360,10 @@ class ClaudeConsoleRelayService {
options = {}
) {
let account = null
const requestId = uuidv4() // 用于并发追踪
let concurrencyAcquired = false
let leaseRefreshInterval = null // 租约刷新定时器
try {
// 获取账户信息
account = await claudeConsoleAccountService.getAccount(accountId)
@@ -320,8 +372,56 @@ class ClaudeConsoleRelayService {
}
logger.info(
`📡 Processing streaming Claude Console API request for key: ${apiKeyData.name || apiKeyData.id}, account: ${account.name} (${accountId})`
`📡 Processing streaming Claude Console API request for key: ${apiKeyData.name || apiKeyData.id}, account: ${account.name} (${accountId}), request: ${requestId}`
)
// 🔒 并发控制：原子性抢占槽位
if (account.maxConcurrentTasks > 0) {
// 先抢占，再检查 - 避免竞态条件
const newConcurrency = Number(
await redis.incrConsoleAccountConcurrency(accountId, requestId, 600)
)
concurrencyAcquired = true
// 检查是否超过限制
if (newConcurrency > account.maxConcurrentTasks) {
// 超限，立即回滚
await redis.decrConsoleAccountConcurrency(accountId, requestId)
concurrencyAcquired = false
logger.warn(
`⚠️ Console account ${account.name} (${accountId}) concurrency limit exceeded: ${newConcurrency}/${account.maxConcurrentTasks} (stream request: ${requestId}, rolled back)`
)
const error = new Error('Console account concurrency limit reached')
error.code = 'CONSOLE_ACCOUNT_CONCURRENCY_FULL'
error.accountId = accountId
throw error
}
logger.debug(
`🔓 Acquired concurrency slot for stream account ${account.name} (${accountId}), current: ${newConcurrency}/${account.maxConcurrentTasks}, request: ${requestId}`
)
// 🔄 启动租约刷新定时器（每5分钟刷新一次，防止长连接租约过期）
leaseRefreshInterval = setInterval(
async () => {
try {
await redis.refreshConsoleAccountConcurrencyLease(accountId, requestId, 600)
logger.debug(
`🔄 Refreshed concurrency lease for stream account ${account.name} (${accountId}), request: ${requestId}`
)
} catch (refreshError) {
logger.error(
`❌ Failed to refresh concurrency lease for account ${accountId}, request: ${requestId}:`,
refreshError.message
)
}
},
5 * 60 * 1000
) // 5分钟刷新一次
}
logger.debug(`🌐 Account API URL: ${account.apiUrl}`)
// 处理模型映射
@@ -373,6 +473,29 @@ class ClaudeConsoleRelayService {
error
)
throw error
} finally {
// 🛑 清理租约刷新定时器
if (leaseRefreshInterval) {
clearInterval(leaseRefreshInterval)
logger.debug(
`🛑 Cleared lease refresh interval for stream account ${account?.name || accountId}, request: ${requestId}`
)
}
// 🔓 并发控制:释放并发槽位
if (concurrencyAcquired) {
try {
await redis.decrConsoleAccountConcurrency(accountId, requestId)
logger.debug(
`🔓 Released concurrency slot for stream account ${account?.name || accountId}, request: ${requestId}`
)
} catch (releaseError) {
logger.error(
`❌ Failed to release concurrency slot for stream account ${accountId}, request: ${requestId}:`,
releaseError.message
)
}
}
}
}