feat: 优化并发控制和移除冗余限制功能

主要改进:
1. 改进并发控制机制
   - 使用 once 代替 on 避免重复监听
   - 监听多个事件确保可靠性(close、finish)
   - 支持客户端断开时立即释放并发槽位

2. 支持非流式请求的客户端断开处理
   - 客户端断开时立即中断上游请求
   - 避免资源浪费和不必要的 API 调用

3. 移除 requestLimit(请求数限制)功能
   - 移除配置和验证逻辑
   - 保留请求统计用于监控分析

4. 移除速率限制(Rate Limit)功能
   - 移除 RATE_LIMIT_* 配置
   - 简化中间件逻辑
   - 避免与并发控制重复

现在系统仅保留:
- Token 使用量限制
- 并发数限制(更精确的资源控制)

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
shaw
2025-07-16 14:40:37 +08:00
parent f9bc2ddb23
commit 567e3b25aa
9 changed files with 62 additions and 108 deletions

View File

@@ -15,7 +15,6 @@ class ApiKeyService {
name = 'Unnamed Key',
description = '',
tokenLimit = config.limits.defaultTokenLimit,
requestLimit = config.limits.defaultRequestLimit,
expiresAt = null,
claudeAccountId = null,
isActive = true,
@@ -33,7 +32,6 @@ class ApiKeyService {
description,
apiKey: hashedKey,
tokenLimit: String(tokenLimit ?? 0),
requestLimit: String(requestLimit ?? 0),
concurrencyLimit: String(concurrencyLimit ?? 0),
isActive: String(isActive),
claudeAccountId: claudeAccountId || '',
@@ -54,7 +52,6 @@ class ApiKeyService {
name: keyData.name,
description: keyData.description,
tokenLimit: parseInt(keyData.tokenLimit),
requestLimit: parseInt(keyData.requestLimit),
concurrencyLimit: parseInt(keyData.concurrencyLimit),
isActive: keyData.isActive === 'true',
claudeAccountId: keyData.claudeAccountId,
@@ -94,15 +91,11 @@ class ApiKeyService {
// 检查使用限制
const usage = await redis.getUsageStats(keyData.id);
const tokenLimit = parseInt(keyData.tokenLimit);
const requestLimit = parseInt(keyData.requestLimit);
if (tokenLimit > 0 && usage.total.tokens >= tokenLimit) {
return { valid: false, error: 'Token limit exceeded' };
}
if (requestLimit > 0 && usage.total.requests >= requestLimit) {
return { valid: false, error: 'Request limit exceeded' };
}
// 更新最后使用时间优化只在实际API调用时更新而不是验证时
// 注意lastUsedAt的更新已移至recordUsage方法中
@@ -116,8 +109,7 @@ class ApiKeyService {
name: keyData.name,
claudeAccountId: keyData.claudeAccountId,
tokenLimit: parseInt(keyData.tokenLimit),
requestLimit: parseInt(keyData.requestLimit),
concurrencyLimit: parseInt(keyData.concurrencyLimit || 0),
concurrencyLimit: parseInt(keyData.concurrencyLimit || 0),
usage
}
};
@@ -136,7 +128,6 @@ class ApiKeyService {
for (const key of apiKeys) {
key.usage = await redis.getUsageStats(key.id);
key.tokenLimit = parseInt(key.tokenLimit);
key.requestLimit = parseInt(key.requestLimit);
key.concurrencyLimit = parseInt(key.concurrencyLimit || 0);
key.currentConcurrency = await redis.getConcurrency(key.id);
key.isActive = key.isActive === 'true';
@@ -159,7 +150,7 @@ class ApiKeyService {
}
// 允许更新的字段
const allowedUpdates = ['name', 'description', 'tokenLimit', 'requestLimit', 'concurrencyLimit', 'isActive', 'claudeAccountId', 'expiresAt'];
const allowedUpdates = ['name', 'description', 'tokenLimit', 'concurrencyLimit', 'isActive', 'claudeAccountId', 'expiresAt'];
const updatedData = { ...keyData };
for (const [field, value] of Object.entries(updates)) {
@@ -240,13 +231,6 @@ class ApiKeyService {
return await redis.getUsageStats(keyId);
}
// 🚦 检查速率限制
async checkRateLimit(keyId, limit = null) {
const rateLimit = limit || config.rateLimit.maxRequests;
const window = Math.floor(config.rateLimit.windowMs / 1000);
return await redis.checkRateLimit(`apikey:${keyId}`, rateLimit, window);
}
// 🧹 清理过期的API Keys
async cleanupExpiredKeys() {

View File

@@ -15,7 +15,9 @@ class ClaudeRelayService {
}
// 🚀 转发请求到Claude API
async relayRequest(requestBody, apiKeyData) {
async relayRequest(requestBody, apiKeyData, clientRequest, clientResponse) {
let upstreamRequest = null;
try {
// 生成会话哈希用于sticky会话
const sessionHash = sessionHelper.generateSessionHash(requestBody);
@@ -34,8 +36,37 @@ class ClaudeRelayService {
// 获取代理配置
const proxyAgent = await this._getProxyAgent(accountId);
// 发送请求到Claude API
const response = await this._makeClaudeRequest(processedBody, accessToken, proxyAgent);
// 设置客户端断开监听器
const handleClientDisconnect = () => {
logger.info('🔌 Client disconnected, aborting upstream request');
if (upstreamRequest && !upstreamRequest.destroyed) {
upstreamRequest.destroy();
}
};
// 监听客户端断开事件
if (clientRequest) {
clientRequest.once('close', handleClientDisconnect);
}
if (clientResponse) {
clientResponse.once('close', handleClientDisconnect);
}
// 发送请求到Claude API传入回调以获取请求对象
const response = await this._makeClaudeRequest(
processedBody,
accessToken,
proxyAgent,
(req) => { upstreamRequest = req; }
);
// 移除监听器(请求成功完成)
if (clientRequest) {
clientRequest.removeListener('close', handleClientDisconnect);
}
if (clientResponse) {
clientResponse.removeListener('close', handleClientDisconnect);
}
// 记录成功的API调用
const inputTokens = requestBody.messages ?
@@ -160,7 +191,7 @@ class ClaudeRelayService {
}
// 🔗 发送请求到Claude API
async _makeClaudeRequest(body, accessToken, proxyAgent) {
async _makeClaudeRequest(body, accessToken, proxyAgent, onRequest) {
return new Promise((resolve, reject) => {
const url = new URL(this.claudeApiUrl);
@@ -207,6 +238,11 @@ class ClaudeRelayService {
}
});
});
// 如果提供了 onRequest 回调,传递请求对象
if (onRequest && typeof onRequest === 'function') {
onRequest(req);
}
req.on('error', (error) => {
logger.error('❌ Claude API request error:', error);