okooooooo/team-api
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix:移除用户和key级别的并发限制

Browse Source
This commit is contained in:
qian committed 2026-06-05 21:11:38 +08:00
1 parent 464e20ef36
commit f0ac2d90d4
8 files changed
+80 -66
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

internal/cmd/cmd.go
+3 -1
View File
@@ -118,7 +118,9 @@ var (
return task.SnapshotHealthScores(ctx)
})
cs.Register("channel_auto_test", "*/5 * * * *", func(ctx context.Context) error {
task.AutoTestChannels(ctx)
if common.Config().GetBool(ctx, "channel_auto_test_enabled") {
task.AutoTestChannels(ctx)
}
return nil
})
cs.Register("model_sunset_check", "0 0 * * *", func(ctx context.Context) error {
internal/logic/billing/ratelimit.go
+30 -41
View File
@@ -25,9 +25,7 @@ type RateLimitConfig struct {
TenantQPS int // 租户级 QPS 限制（默认 1000）
UserQPS int // 用户级 QPS 限制（默认 100）
KeyQPS int // Key 级 QPS 限制（默认 60）
TenantConc int // 租户级并发限制（默认 50）
UserConc int // 用户级并发限制（默认 10）
KeyConc int // Key 级并发限制（默认 5）
TenantConc int // 租户级并发限制（默认 0，不限制）
}
// DefaultRateLimitConfig 默认限流配置
@@ -36,9 +34,7 @@ var DefaultRateLimitConfig = RateLimitConfig{
TenantQPS: 1000,
UserQPS: 100,
KeyQPS: 60,
TenantConc: 50,
UserConc: 10,
KeyConc: 5,
TenantConc: 0,
}
// LoadRateLimitConfig loads rate limit config from ConfigService, falling back to defaults.
@@ -174,20 +170,20 @@ func CheckRateLimit(ctx context.Context, config RateLimitConfig, tenantID, userI
}
// ---------------------------------------------------------------------------
// 并发限流：单个 Lua 脚本完成 4 级获取 + 失败回滚
// 并发限流：单个 Lua 脚本完成 2 级获取 + 失败回滚（租户 + 模型）
// ---------------------------------------------------------------------------
// acquireConcurrentLua 一次 EVAL 完成 4 级并发许可获取。
// KEYS[1..4]: tenant/model/user/key 的 Redis key
// ARGV[1..4]: 各级并发限制（0 表示不检查该级）
// ARGV[5]: EXPIRE 秒数
// acquireConcurrentLua 一次 EVAL 完成 2 级并发许可获取（租户 + 模型）。
// KEYS[1..2]: tenant/model 的 Redis key
// ARGV[1..2]: 各级并发限制（0 表示不检查该级）
// ARGV[3]: EXPIRE 秒数
//
// 逐级尝试获取，任一级失败则回滚已获取的许可。
// 返回 1=成功, 0=失败。
var acquireConcurrentLua = `
local keys = {KEYS[1], KEYS[2], KEYS[3], KEYS[4]}
local limits = {tonumber(ARGV[1]), tonumber(ARGV[2]), tonumber(ARGV[3]), tonumber(ARGV[4])}
local expire = tonumber(ARGV[5])
local keys = {KEYS[1], KEYS[2]}
local limits = {tonumber(ARGV[1]), tonumber(ARGV[2])}
local expire = tonumber(ARGV[3])
local acquired = {}
local function tryAcquire(i)
@@ -212,16 +208,14 @@ end
if not tryAcquire(1) then return 0 end
if not tryAcquire(2) then return 0 end
if not tryAcquire(3) then return 0 end
if not tryAcquire(4) then return 0 end
return 1
`
// releaseConcurrentLua 一次 EVAL 释放 4 级并发许可。
// releaseConcurrentLua 一次 EVAL 释放 2 级并发许可（租户 + 模型）。
// ARGV[1] = expire 秒数（用于 SET 后恢复 TTL）
var releaseConcurrentLua = `
local expire = tonumber(ARGV[1])
for i = 1, 4 do
for i = 1, 2 do
local v = redis.call("DECR", KEYS[i])
if v < 0 then
redis.call("SET", KEYS[i], 0)
@@ -231,23 +225,21 @@ end
return 1
`
// AcquireConcurrent 获取并发许可（四级：租户→模型→用户→Key）
// AcquireConcurrent 获取并发许可（两级：租户→模型）
// 租户级并发限制优先使用 tnt_tenants.max_concurrency，0 表示不限；
// 模型级并发限制使用 mdl_tenant_models.max_concurrency，NULL 表示不限；
// 若未设置则回退到全局配置 tenant_concurrency_limit。
func AcquireConcurrent(ctx context.Context, config RateLimitConfig, tenantID, userID, apiKeyID int64, modelName string) bool {
tenantKey := fmt.Sprintf("conc:tenant:%d", tenantID)
modelKey := fmt.Sprintf("conc:tenant:%d:model:%s", tenantID, modelName)
userKey := fmt.Sprintf("conc:user:%d", userID)
keyKey := fmt.Sprintf("conc:key:%d", apiKeyID)
// 获取租户级和模型级并发限制（含缓存，可能查 DB）
tenantLimit := getTenantConcurrencyLimit(ctx, tenantID, config.TenantConc)
tenantLimit := getTenantConcurrencyLimit(ctx, tenantID)
modelLimit := getModelConcurrencyLimit(ctx, tenantID, modelName)
result, err := g.Redis().Do(ctx, "EVAL", acquireConcurrentLua, 4,
tenantKey, modelKey, userKey, keyKey,
tenantLimit, modelLimit, config.UserConc, config.KeyConc,
result, err := g.Redis().Do(ctx, "EVAL", acquireConcurrentLua, 2,
tenantKey, modelKey,
tenantLimit, modelLimit,
300) // EXPIRE 秒数
if err != nil {
return true // Redis 不可用，允许通过
@@ -256,23 +248,21 @@ func AcquireConcurrent(ctx context.Context, config RateLimitConfig, tenantID, us
return result.Int() == 1
}
// ReleaseConcurrent 释放并发许可（含模型级）
// 单次 EVAL 释放所有 4 个 key，替代原来的 4 次 DECR。
// ReleaseConcurrent 释放并发许可（租户 + 模型两级）
// 单次 EVAL 释放所有 2 个 key。
func ReleaseConcurrent(ctx context.Context, tenantID, userID, apiKeyID int64, modelName string) {
tenantKey := fmt.Sprintf("conc:tenant:%d", tenantID)
modelKey := fmt.Sprintf("conc:tenant:%d:model:%s", tenantID, modelName)
userKey := fmt.Sprintf("conc:user:%d", userID)
keyKey := fmt.Sprintf("conc:key:%d", apiKeyID)
g.Redis().Do(ctx, "EVAL", releaseConcurrentLua, 4,
tenantKey, modelKey, userKey, keyKey,
g.Redis().Do(ctx, "EVAL", releaseConcurrentLua, 2,
tenantKey, modelKey,
300) // EXPIRE 秒数
}
// getTenantConcurrencyLimit 获取租户并发上限。
// 优先从 Redis 缓存读取 tnt_tenants.max_concurrency，缓存 300 秒；
// 缓存未命中则查询数据库并回填。0 表示不限，回退到全局默认值。
func getTenantConcurrencyLimit(ctx context.Context, tenantID int64, defaultLimit int) int {
// 缓存未命中则查询数据库并回填。0 表示不限。
func getTenantConcurrencyLimit(ctx context.Context, tenantID int64) int {
cacheKey := fmt.Sprintf("tenant:conc_limit:%d", tenantID)
// 尝试从 Redis 缓存读取
@@ -295,23 +285,23 @@ func getTenantConcurrencyLimit(ctx context.Context, tenantID int64, defaultLimit
Fields("max_concurrency").
Scan(&maxConc)
if err != nil {
// 查询失败，使用全局默认值
return defaultLimit
// 查询失败，不限制
return 0
}
// NULL 表示跟随等级配置
if maxConc == nil {
_, effectiveConc, err := GetTenantEffectiveLimits(ctx, tenantID)
if err != nil {
return defaultLimit
return 0
}
if effectiveConc > 0 {
g.Redis().Do(ctx, "SET", cacheKey, effectiveConc, "EX", 300)
return effectiveConc
}
// 等级配置也是 0（不限），缓存 -1 标记，回退到全局默认
// 等级配置也是 0（不限），缓存 -1 标记
g.Redis().Do(ctx, "SET", cacheKey, -1, "EX", 300)
return defaultLimit
return 0
}
if *maxConc > 0 {
@@ -320,10 +310,9 @@ func getTenantConcurrencyLimit(ctx context.Context, tenantID int64, defaultLimit
return *maxConc
}
// 数据库中为 0（不限），缓存 -1 标记，回退到全局默认
// 数据库中为 0（不限），缓存 -1 标记
g.Redis().Do(ctx, "SET", cacheKey, -1, "EX", 300)
return defaultLimit
return 0
}
// getModelConcurrencyLimit 获取租户模型级并发上限。
internal/logic/billing/ratelimit_test.go
+2 -8
View File
@@ -19,14 +19,8 @@ func TestRateLimitConfig(t *testing.T) {
if cfg.KeyQPS != 60 {
t.Errorf("expected KeyQPS 60, got %d", cfg.KeyQPS)
}
if cfg.TenantConc != 50 {
t.Errorf("expected TenantConc 50, got %d", cfg.TenantConc)
}
if cfg.UserConc != 10 {
t.Errorf("expected UserConc 10, got %d", cfg.UserConc)
}
if cfg.KeyConc != 5 {
t.Errorf("expected KeyConc 5, got %d", cfg.KeyConc)
if cfg.TenantConc != 0 {
t.Errorf("expected TenantConc 0, got %d", cfg.TenantConc)
}
}
internal/logic/common/config.go
+2 -1
View File
@@ -243,7 +243,8 @@ func (s *ConfigService) UpdateCategory(ctx context.Context, category string, val
for key, value := range values {
def, ok := defMap[key]
if !ok {
return fmt.Errorf("配置项 %s 不属于分类 %s", key, category)
// 注册表中已移除的配置项，跳过（数据库中可能残留旧行）
continue
}
if def.Sensitive && value == "******" {
continue
internal/logic/common/settings_registry.go
+11 -7
View File
@@ -158,15 +158,15 @@ var Registry = []SettingDef{
// ── Performance ──
{Key: "global_qps_limit", Type: SettingTypeInt, Default: "10000", Category: "performance",
Label: "系统级 QPS 上限", Validation: "min:100,max:1000000"},
Label: "系统级 QPS 上限", Validation: "min:0"},
{Key: "tenant_qps_limit", Type: SettingTypeInt, Default: "1000", Category: "performance",
Label: "租户级 QPS 上限", Validation: "min:10,max:100000"},
Label: "租户级 QPS 上限", Validation: "min:0"},
{Key: "user_qps_limit", Type: SettingTypeInt, Default: "100", Category: "performance",
Label: "用户级 QPS 上限", Validation: "min:1,max:10000"},
Label: "用户级 QPS 上限", Validation: "min:0"},
{Key: "key_qps_limit", Type: SettingTypeInt, Default: "60", Category: "performance",
Label: "Key 级 QPS 上限", Validation: "min:1,max:10000"},
{Key: "tenant_concurrency_limit", Type: SettingTypeInt, Default: "50", Category: "performance",
Label: "租户级并发上限", Validation: "min:1,max:10000"},
Label: "Key 级 QPS 上限", Validation: "min:0"},
{Key: "tenant_concurrency_limit", Type: SettingTypeInt, Default: "0", Category: "performance",
Label: "租户级并发上限", Validation: "min:0"},
{Key: "request_timeout_seconds", Type: SettingTypeInt, Default: "120", Category: "performance",
Label: "请求超时(秒)", Validation: "min:5,max:600"},
{Key: "batch_write_size", Type: SettingTypeInt, Default: "64", Category: "performance",
@@ -187,7 +187,7 @@ var Registry = []SettingDef{
Label: "批量写入间隔(ms)", Validation: "min:100,max:30000",
Description: "日志批量写入的间隔时间"},
{Key: "global_concurrency_limit", Type: SettingTypeInt, Default: "1000", Category: "performance",
Label: "系统级并发上限", Validation: "min:10,max:100000"},
Label: "系统级并发上限", Validation: "min:0"},
{Key: "pprof_port", Type: SettingTypeInt, Default: "0", Category: "performance",
Label: "pprof 端口", Validation: "min:0,max:65535",
Description: "性能分析端口，0=关闭"},
@@ -204,6 +204,10 @@ var Registry = []SettingDef{
Label: "拦截提示", Description: "block 策略下返回给客户端的提示信息"},
// ── Channel ──
{Key: "channel_auto_test_enabled", Type: SettingTypeBool, Default: "true", Category: "channel",
Label: "渠道自动探测", Description: "定期向活跃渠道发送测试请求，检测连通性并更新健康度（会消耗少量 Token）"},
{Key: "channel_auto_test_recovery_enabled", Type: SettingTypeBool, Default: "true", Category: "channel",
Label: "自动恢复探测", Description: "定期测试已自动禁用的渠道，测试通过则恢复启用（依赖自动探测开启）"},
{Key: "channel_auto_disable_enabled", Type: SettingTypeBool, Default: "false", Category: "channel",
Label: "渠道自动禁用", Description: "连续失败达到阈值时自动禁用渠道"},
{Key: "channel_auto_disable_threshold", Type: SettingTypeInt, Default: "5", Category: "channel",
internal/logic/task/channel_auto_test_cron.go
+5 -2
View File
@@ -9,6 +9,7 @@ import (
v1 "github.com/qianfree/team-api/api/admin/v1"
"github.com/qianfree/team-api/internal/dao"
"github.com/qianfree/team-api/internal/logic/admin"
"github.com/qianfree/team-api/internal/logic/common"
do "github.com/qianfree/team-api/internal/model/do"
)
@@ -58,8 +59,10 @@ func AutoTestChannels(ctx context.Context) {
}
}
// 2. 尝试恢复自动禁用的渠道
testAndRecoverDisabledChannels(ctx)
// 2. 尝试恢复自动禁用的渠道（受独立开关控制）
if common.Config().GetBool(ctx, "channel_auto_test_recovery_enabled") {
testAndRecoverDisabledChannels(ctx)
}
}
// testAndRecoverDisabledChannels 测试自动禁用的渠道，测试通过则恢复
web/admin/src/views/system/settings/ChannelTab.vue
+21
View File
@@ -5,6 +5,27 @@ const values = useFormValues()
<template>
<div class="tab-content">
<!-- 自动探测 -->
<div class="section">
<div class="section-title">自动探测</div>
<div class="switch-row">
<span class="switch-label">渠道自动探测</span>
<span class="switch-desc">定期向活跃渠道发送测试请求，检测连通性并更新健康度（会消耗少量 Token）</span>
<ASwitch
:model-value="values['channel_auto_test_enabled'] === 'true' || values['channel_auto_test_enabled'] === '1'"
@change="(v: string | number | boolean) => values['channel_auto_test_enabled'] = String(v)"
/>
</div>
<div class="switch-row" style="margin-top: 12px">
<span class="switch-label">自动恢复探测</span>
<span class="switch-desc">定期测试已自动禁用的渠道，测试通过则恢复启用（依赖自动探测开启）</span>
<ASwitch
:model-value="values['channel_auto_test_recovery_enabled'] === 'true' || values['channel_auto_test_recovery_enabled'] === '1'"
@change="(v: string | number | boolean) => values['channel_auto_test_recovery_enabled'] = String(v)"
/>
</div>
</div>
<!-- 自动禁用 -->
<div class="section">
<div class="section-title">自动禁用</div>
web/admin/src/views/system/settings/PerformanceTab.vue
+6 -6
View File
@@ -17,19 +17,19 @@ function set(key: string, v: number | undefined): void {
<div class="group-title">QPS 限制</div>
<div class="compact-row">
<label>系统级</label>
<AInputNumber :model-value="num('global_qps_limit')" @change="(v: number | undefined) => set('global_qps_limit', v)" :min="100" :max="1000000" />
<AInputNumber :model-value="num('global_qps_limit')" @change="(v: number | undefined) => set('global_qps_limit', v)" :min="0" />
</div>
<div class="compact-row">
<label>租户级</label>
<AInputNumber :model-value="num('tenant_qps_limit')" @change="(v: number | undefined) => set('tenant_qps_limit', v)" :min="10" :max="100000" />
<AInputNumber :model-value="num('tenant_qps_limit')" @change="(v: number | undefined) => set('tenant_qps_limit', v)" :min="0" />
</div>
<div class="compact-row">
<label>用户级</label>
<AInputNumber :model-value="num('user_qps_limit')" @change="(v: number | undefined) => set('user_qps_limit', v)" :min="1" :max="10000" />
<AInputNumber :model-value="num('user_qps_limit')" @change="(v: number | undefined) => set('user_qps_limit', v)" :min="0" />
</div>
<div class="compact-row">
<label>Key 级</label>
<AInputNumber :model-value="num('key_qps_limit')" @change="(v: number | undefined) => set('key_qps_limit', v)" :min="1" :max="10000" />
<AInputNumber :model-value="num('key_qps_limit')" @change="(v: number | undefined) => set('key_qps_limit', v)" :min="0" />
</div>
</div>
@@ -38,11 +38,11 @@ function set(key: string, v: number | undefined): void {
<div class="group-title">并发限制</div>
<div class="compact-row">
<label>系统级上限</label>
<AInputNumber :model-value="num('global_concurrency_limit')" @change="(v: number | undefined) => set('global_concurrency_limit', v)" :min="10" :max="100000" />
<AInputNumber :model-value="num('global_concurrency_limit')" @change="(v: number | undefined) => set('global_concurrency_limit', v)" :min="0" />
</div>
<div class="compact-row">
<label>租户级上限</label>
<AInputNumber :model-value="num('tenant_concurrency_limit')" @change="(v: number | undefined) => set('tenant_concurrency_limit', v)" :min="1" :max="10000" />
<AInputNumber :model-value="num('tenant_concurrency_limit')" @change="(v: number | undefined) => set('tenant_concurrency_limit', v)" :min="0" />
</div>
</div>