From 43f7d30e46995cef663dcd6b153ba02981bf5c34 Mon Sep 17 00:00:00 2001 From: yixiong Date: Thu, 9 Jul 2026 08:39:43 +0800 Subject: [PATCH] =?UTF-8?q?feat(t9):=20=E6=8C=89=E7=A7=9F=E6=88=B7?= =?UTF-8?q?=E4=BB=A4=E7=89=8C=E6=A1=B6=E9=99=90=E6=B5=81=EF=BC=88=E5=87=BA?= =?UTF-8?q?=E7=AB=99=E8=85=BE=E8=AE=AFAPI=E9=85=8D=E9=A2=9D=E4=BF=9D?= =?UTF-8?q?=E6=8A=A4=EF=BC=8C=E5=AF=B9=E9=BD=90=E8=AE=BE=E8=AE=A13.2?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - RateLimiter(新): Redis+Lua 原子令牌桶，租户桶(quota_im_qps)+全局桶(globalImQps)， 任一超限抛 BizException(429)；Redis 异常 fail-open 放行 - RedisService: +eval(Lua) 通用脚本执行（令牌桶复合操作需原子） - TencentImClient.callApi: 开头 checkApiLimit，resolveTenantQps 查 quota_im_qps(Caffeine 60s) - 联调：sa quota=2，8 并发→2 放行+6 限流，被拒不调腾讯（api_call_log 仅 2 条） Co-Authored-By: Claude --- .../java/com/imutil/common/RateLimiter.java | 97 +++++++++++++++++++ .../java/com/imutil/common/RedisService.java | 14 +++ .../com/imutil/tencent/TencentImClient.java | 53 ++++++++++ 3 files changed, 164 insertions(+) create mode 100644 src/main/java/com/imutil/common/RateLimiter.java diff --git a/src/main/java/com/imutil/common/RateLimiter.java b/src/main/java/com/imutil/common/RateLimiter.java new file mode 100644 index 0000000..6414117 --- /dev/null +++ b/src/main/java/com/imutil/common/RateLimiter.java @@ -0,0 +1,97 @@ +package com.imutil.common; + +import lombok.extern.slf4j.Slf4j; +import org.noear.solon.annotation.Component; +import org.noear.solon.annotation.Inject; + +import java.util.List; + +/** + * 按租户令牌桶限流（出站腾讯 API 配额保护，对齐设计 3.2） + *

+ * 合用 SDKAppID 后所有租户瓜分腾讯侧共享 QPS，本工具作为唯一出站入口， + * 在调腾讯 API 前做两层令牌桶限流： + * 1. 租户桶（tenant.quota_im_qps，缺省 defaultImQps）—— 防单租户突发； + * 2. 全局桶（globalImQps，腾讯限制的 80%）—— 兜底共享配额。 + * 任一桶令牌耗尽即拒绝（抛 {@link BizException} 429）。 + *

+ * 令牌桶用 Redis + Lua 原子实现，多节点一致；Redis 异常时 fail-open（放行，避免拖垮业务）。 + * 容量=填充速率=QPS，即严格每秒 QPS 次（不为突发额外放大容量，以稳守腾讯配额）。 + * + * @author imutil + */ +@Slf4j +@Component +public class RateLimiter { + + /** 令牌桶 Lua：按时间填充令牌、扣减，返回 1=放行 0=超限 */ + private static final String TOKEN_BUCKET_LUA = + "local key = KEYS[1] " + + "local capacity = tonumber(ARGV[1]) " + + "local rate = tonumber(ARGV[2]) " + + "local now = tonumber(ARGV[3]) " + + "local requested = tonumber(ARGV[4]) " + + "local tokens = tonumber(redis.call('HGET', key, 'tokens')) " + + "local last = tonumber(redis.call('HGET', key, 'last')) " + + "if tokens == nil then tokens = capacity end " + + "if last == nil then last = now end " + + "local delta = math.max(0, now - last) / 1000.0 * rate " + + "tokens = math.min(capacity, tokens + delta) " + + "local allowed = 0 " + + "if tokens >= requested then tokens = tokens - requested; allowed = 1 end " + + "redis.call('HMSET', key, 'tokens', tokens, 'last', now) " + + "redis.call('EXPIRE', key, math.ceil(capacity / rate) + 60) " + + "return allowed"; + + @Inject("${imutil.ratelimit.globalImQps:200}") + private int globalImQps; + + @Inject + private RedisService redisService; + + /** + * 出站腾讯 API 限流检查：租户桶 + 全局桶，任一超限抛 BizException(429) + * + * @param tenantId 当前租户（无上下文为 system） + * @param tenantQps 该租户配额（QPS），由调用方解析（tenant.quota_im_qps 或默认） + */ + public void checkApiLimit(String tenantId, int tenantQps) { + // 1. 租户桶：防单租户突发打爆共享配额 + if (!tryAcquire("rl:tenant:" + tenantId, tenantQps)) { + log.warn("租户IM API配额超限 tenant={} qps={}", tenantId, tenantQps); + throw new BizException(429, "租户[" + tenantId + "] IM API 配额超限，请稍后重试"); + } + // 2. 全局桶：兜底共享 SDKAppID 总配额（腾讯限制的 80%） + if (!tryAcquire("rl:global", globalImQps)) { + log.warn("全局IM API配额超限 globalQps={}", globalImQps); + throw new BizException(429, "全局 IM API 配额超限，请稍后重试"); + } + } + + /** + * 令牌桶尝试获取 1 个令牌（capacity=rate=qps，严格每秒 qps 次） + *

+ * Redis 异常时 fail-open 放行：限流是保护措施，不可因 Redis 故障拖垮业务调用。 + * + * @param key 桶键 + * @param qps 每秒允许次数（容量与填充速率均取此值）；<=0 表示不限流 + * @return true=放行 false=超限 + */ + private boolean tryAcquire(String key, int qps) { + if (qps <= 0) { + return true; + } + try { + long now = System.currentTimeMillis(); + Object r = redisService.eval(TOKEN_BUCKET_LUA, + List.of(key), + List.of(String.valueOf(qps), String.valueOf(qps), String.valueOf(now), "1")); + // Lua 返回 number，Jedis 转 Long + long allowed = (r instanceof Number) ? ((Number) r).longValue() : Long.parseLong(String.valueOf(r)); + return allowed == 1L; + } catch (Exception e) { + log.warn("限流Redis异常，fail-open放行 key={} : {}", key, e.getMessage()); + return true; + } + } +} diff --git a/src/main/java/com/imutil/common/RedisService.java b/src/main/java/com/imutil/common/RedisService.java index 93f806a..0d7e948 100644 --- a/src/main/java/com/imutil/common/RedisService.java +++ b/src/main/java/com/imutil/common/RedisService.java @@ -139,4 +139,18 @@ public class RedisService { return j.ttl(key); } } + + /** + * 执行 Lua 脚本（原子操作，用于令牌桶限流等复合场景） + * + * @param script Lua 脚本 + * @param keys KEYS 列表 + * @param args ARGV 列表 + * @return 脚本返回值（Jedis 按类型转 Long/String/List） + */ + public Object eval(String script, java.util.List keys, java.util.List args) { + try (Jedis j = jedisPool.getResource()) { + return j.eval(script, keys, args); + } + } } diff --git a/src/main/java/com/imutil/tencent/TencentImClient.java b/src/main/java/com/imutil/tencent/TencentImClient.java index 01af1f6..18fc1ce 100644 --- a/src/main/java/com/imutil/tencent/TencentImClient.java +++ b/src/main/java/com/imutil/tencent/TencentImClient.java @@ -2,9 +2,13 @@ package com.imutil.tencent; import com.imutil.common.Httpx; import com.imutil.common.Jsons; +import com.imutil.common.LocalCache; +import com.imutil.common.RateLimiter; import com.imutil.common.TenantContext; import com.imutil.entity.ApiCallLog; +import com.imutil.entity.Tenant; import com.imutil.mapper.ApiCallLogMapper; +import com.imutil.service.TenantService; import lombok.extern.slf4j.Slf4j; import org.noear.snack4.ONode; import org.noear.solon.annotation.Component; @@ -49,6 +53,18 @@ public class TencentImClient { @Inject private ApiCallLogMapper apiCallLogMapper; + @Inject("${imutil.ratelimit.defaultImQps:50}") + private int defaultImQps; + + @Inject + private RateLimiter rateLimiter; + + @Inject + private TenantService tenantService; + + @Inject + private LocalCache localCache; + /** * 生成管理员 UserSig（长效，用于调后台 API） */ @@ -68,6 +84,10 @@ public class TencentImClient { String tenantId = TenantContext.get(); String tid = (tenantId == null || tenantId.isEmpty()) ? "system" : tenantId; + // 出站腾讯 API 限流：按租户配额 + 全局共享配额（对齐设计 3.2），超限抛 429 + // system（无租户上下文）用默认配额；补拉等后台任务超限由调用方 catch 跳过本轮 + rateLimiter.checkApiLimit(tid, resolveTenantQps(tid)); + String result; try { String adminSig = genAdminSig(); @@ -110,6 +130,39 @@ public class TencentImClient { } } + /** + * 解析租户 IM API 配额：tenant.quota_im_qps，缺省 defaultImQps；本地缓存 60s + *

+ * 配额变更（管理后台改 quota_im_qps）最长 60s 生效；system（无租户上下文）用默认配额。 + * + * @param tid 租户ID + * @return 该租户 IM API QPS 配额 + */ + private int resolveTenantQps(String tid) { + if (tid == null || tid.isEmpty() || "system".equals(tid)) { + return defaultImQps; + } + String cacheKey = "tenantQps:" + tid; + String cached = localCache.get(cacheKey); + if (cached != null) { + try { + return Integer.parseInt(cached); + } catch (NumberFormatException ignore) { + } + } + int qps = defaultImQps; + try { + Tenant t = tenantService.getById(tid); + if (t != null && t.getQuotaImQps() != null && t.getQuotaImQps() > 0) { + qps = t.getQuotaImQps(); + } + } catch (Exception e) { + log.warn("查询租户配额失败 tenant={}，用默认 {} : {}", tid, defaultImQps, e.getMessage()); + } + localCache.put(cacheKey, String.valueOf(qps)); + return qps; + } + /** * 导入账号（创建 IM 用户，幂等：已存在亦返回 OK） *