|
|
@@ -1,5 +1,7 @@
|
|
|
package com.pharmacopoeia.security;
|
|
|
|
|
|
+import com.github.benmanes.caffeine.cache.Cache;
|
|
|
+import com.github.benmanes.caffeine.cache.Caffeine;
|
|
|
import com.pharmacopoeia.config.RateLimitProperties;
|
|
|
import jakarta.servlet.FilterChain;
|
|
|
import org.jetbrains.annotations.NotNull;
|
|
|
@@ -8,40 +10,34 @@ import com.pharmacopoeia.util.IpUtils;
|
|
|
import jakarta.servlet.ServletException;
|
|
|
import jakarta.servlet.http.HttpServletRequest;
|
|
|
import jakarta.servlet.http.HttpServletResponse;
|
|
|
-import org.slf4j.Logger;
|
|
|
-import org.slf4j.LoggerFactory;
|
|
|
-import org.springframework.scheduling.annotation.Scheduled;
|
|
|
import org.springframework.stereotype.Component;
|
|
|
import org.springframework.web.filter.OncePerRequestFilter;
|
|
|
|
|
|
import java.io.IOException;
|
|
|
-import java.util.Iterator;
|
|
|
-import java.util.Map;
|
|
|
-import java.util.concurrent.ConcurrentHashMap;
|
|
|
+import java.time.Duration;
|
|
|
+import java.util.ArrayDeque;
|
|
|
+import java.util.Deque;
|
|
|
|
|
|
/**
|
|
|
- * 用户侧限流:仅 chat POST 接口,按 IP 固定窗口(60s)计数。
|
|
|
- * 窗口内累加;超过 perMinute 继续拦;窗口时间过去后计数归零、自动恢复。
|
|
|
- * 超限返回 429,不进入 Controller 层,不写缓存和 DB。
|
|
|
+ * 用户/系统侧限流(按 IP 滑动窗口 60s):
|
|
|
+ * - chat POST(/chat/*):perMinute(默认 60)—— 问答不能无限刷
|
|
|
+ * - 埋点写入 POST(/api/v1/analytics/events):analyticsPerMinute(默认 600,多系统调用给大)
|
|
|
+ * 时间戳队列 + Caffeine expireAfterAccess(60s) 自动淘汰空闲桶,无需手动清理。
|
|
|
+ * 超限返回 429,不进入 Controller 层。
|
|
|
*/
|
|
|
@Component
|
|
|
public class RateLimitFilter extends OncePerRequestFilter {
|
|
|
|
|
|
- private static final Logger log = LoggerFactory.getLogger(RateLimitFilter.class);
|
|
|
private static final long WINDOW_MS = 60_000L;
|
|
|
|
|
|
private final RateLimitProperties props;
|
|
|
- private final Map<String, Window> buckets = new ConcurrentHashMap<>();
|
|
|
+ private final Cache<String, Deque<Long>> chatBuckets;
|
|
|
+ private final Cache<String, Deque<Long>> analyticsBuckets;
|
|
|
|
|
|
public RateLimitFilter(RateLimitProperties props) {
|
|
|
this.props = props;
|
|
|
- }
|
|
|
-
|
|
|
- /** 每 IP 一个窗口:起始时间 + 计数。 */
|
|
|
- private static final class Window {
|
|
|
- volatile long start;
|
|
|
- volatile int count;
|
|
|
- Window(long start) { this.start = start; }
|
|
|
+ this.chatBuckets = Caffeine.newBuilder().expireAfterAccess(Duration.ofSeconds(60)).build();
|
|
|
+ this.analyticsBuckets = Caffeine.newBuilder().expireAfterAccess(Duration.ofSeconds(60)).build();
|
|
|
}
|
|
|
|
|
|
@Override
|
|
|
@@ -51,9 +47,17 @@ public class RateLimitFilter extends OncePerRequestFilter {
|
|
|
String ip = IpUtils.getClientIp(request);
|
|
|
MDC.put("ip", ip);
|
|
|
try {
|
|
|
- boolean isChatApi = request.getRequestURI().contains("/chat/")
|
|
|
- && "POST".equalsIgnoreCase(request.getMethod());
|
|
|
- if (isChatApi && !allow(ip)) {
|
|
|
+ String uri = request.getRequestURI();
|
|
|
+ boolean post = "POST".equalsIgnoreCase(request.getMethod());
|
|
|
+
|
|
|
+ boolean limited = false;
|
|
|
+ if (post && uri.contains("/chat/")) {
|
|
|
+ limited = !allow(ip, chatBuckets, props.getPerMinute());
|
|
|
+ } else if (post && uri.contains("/api/v1/analytics/events")) {
|
|
|
+ limited = !allow(ip, analyticsBuckets, props.getAnalyticsPerMinute());
|
|
|
+ }
|
|
|
+
|
|
|
+ if (limited) {
|
|
|
writeRateLimitError(response);
|
|
|
return;
|
|
|
}
|
|
|
@@ -63,47 +67,23 @@ public class RateLimitFilter extends OncePerRequestFilter {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- private boolean allow(String ip) {
|
|
|
- int perMinute = props.getPerMinute();
|
|
|
+ private boolean allow(String ip, Cache<String, Deque<Long>> buckets, int perMinute) {
|
|
|
+ if (perMinute <= 0) return true; // 0/负 = 不限
|
|
|
long now = System.currentTimeMillis();
|
|
|
- Window w = buckets.computeIfAbsent(ip, k -> new Window(now));
|
|
|
- synchronized (w) {
|
|
|
- if (now - w.start >= WINDOW_MS) {
|
|
|
- // 新窗口:归零计数、刷新起始时间 → 自动恢复
|
|
|
- w.start = now;
|
|
|
- w.count = 1;
|
|
|
- } else {
|
|
|
- w.count++;
|
|
|
+ Deque<Long> dq = buckets.get(ip, k -> new ArrayDeque<>());
|
|
|
+ synchronized (dq) {
|
|
|
+ long cutoff = now - WINDOW_MS;
|
|
|
+ while (!dq.isEmpty() && dq.peekFirst() < cutoff) {
|
|
|
+ dq.pollFirst();
|
|
|
}
|
|
|
- if (w.count > perMinute) {
|
|
|
- w.count--; // 超限不再累加,避免计数无限上涨;窗口结束前继续拦
|
|
|
+ if (dq.size() >= perMinute) {
|
|
|
return false;
|
|
|
}
|
|
|
+ dq.addLast(now);
|
|
|
return true;
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- /** 每 2 分钟清理一次无活动 IP 的桶,防 map 无限增长 */
|
|
|
- @Scheduled(fixedRate = 120_000)
|
|
|
- public void cleanupStaleBuckets() {
|
|
|
- long now = System.currentTimeMillis();
|
|
|
- int removed = 0;
|
|
|
- Iterator<Map.Entry<String, Window>> it = buckets.entrySet().iterator();
|
|
|
- while (it.hasNext()) {
|
|
|
- Map.Entry<String, Window> e = it.next();
|
|
|
- Window w = e.getValue();
|
|
|
- synchronized (w) {
|
|
|
- if (now - w.start >= WINDOW_MS * 2) {
|
|
|
- it.remove();
|
|
|
- removed++;
|
|
|
- }
|
|
|
- }
|
|
|
- }
|
|
|
- if (removed > 0) {
|
|
|
- log.debug("[rateLimit] 清理 {} 个过期 IP 窗口", removed);
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
private static void writeRateLimitError(HttpServletResponse response) throws IOException {
|
|
|
response.setStatus(429);
|
|
|
response.setContentType("application/json;charset=UTF-8");
|