Spring AOP中如何实现带并发控制的@RateLimit自定义注解？

题目

Spring AOP中如何实现带并发控制的@RateLimit自定义注解？

信息

• 类型：问答
• 难度：⭐⭐⭐

考点

自定义注解实现,环绕通知编程,并发控制,分布式限流,AOP底层原理

快速回答

实现带并发控制的@RateLimit注解需要：

• 定义@RateLimit注解包含限流参数
• 使用@Around环绕通知实现限流逻辑
• 通过ConcurrentHashMap+AtomicInteger实现单机计数器
• 使用Semaphore或RateLimiter控制并发
• 结合Redis实现分布式限流
• 处理限流触发时的降级响应

1. 核心需求与难点

在分布式系统中实现方法级限流，要求：

• 通过自定义注解声明式控制
• 支持每秒请求数(QPS)和并发线程数控制
• 线程安全且高性能
• 支持分布式环境扩展

2. 实现步骤与代码示例

2.1 定义注解

@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface RateLimit {
    int value() default 10; // 默认QPS=10
    int concurrency() default 5; // 默认并发数=5
}

2.2 环绕通知实现（单机版）

@Aspect
@Component
public class RateLimitAspect {
    // QPS计数器
    private final Map<String, AtomicInteger> counterMap = new ConcurrentHashMap<>();
    // 并发控制器
    private final Map<String, Semaphore> semaphoreMap = new ConcurrentHashMap<>();

    @Around("@annotation(rateLimit)")
    public Object doRateLimit(ProceedingJoinPoint pjp, RateLimit rateLimit) throws Throwable {
        String methodKey = pjp.getSignature().toLongString();

        // 初始化并发控制器
        semaphoreMap.computeIfAbsent(methodKey, k -> 
            new Semaphore(rateLimit.concurrency()));

        // 获取许可（非阻塞版）
        if (!semaphoreMap.get(methodKey).tryAcquire()) {
            throw new RateLimitException("并发数超过限制");
        }

        try {
            // QPS检查
            AtomicInteger counter = counterMap.computeIfAbsent(methodKey, 
                k -> new AtomicInteger(0));

            if (counter.incrementAndGet() > rateLimit.value()) {
                throw new RateLimitException("QPS超过限制");
            }

            // 执行目标方法
            return pjp.proceed();
        } finally {
            semaphoreMap.get(methodKey).release();
            // 使用定时器重置计数器（实际需用ScheduledExecutor）
        }
    }
}

3. 关键难点解决方案

3.1 时间窗口重置

• 问题：QPS计数器需要每秒重置
• 方案：使用ScheduledExecutorService定时重置
• 优化：Guava的RateLimiter或Bucket4j

3.2 分布式限流

// Redis Lua脚本实现原子操作
String luaScript = "local current = redis.call('incr', KEYS[1])\n" +
                   "if current == 1 then\n" +
                   "    redis.call('expire', KEYS[1], 1)\n" +
                   "end\n" +
                   "return current";

// 执行脚本
Long count = redisTemplate.execute(
    new DefaultRedisScript<>(luaScript, Long.class),
    Collections.singletonList(methodKey)
);

if (count > rateLimit.value()) {
    throw new RateLimitException();
}

3.3 性能优化

• 使用LongAdder替代AtomicInteger减少CAS冲突
• 为不同方法Key做分片锁
• 异步更新计数避免阻塞主线程

4. 最佳实践

• 降级策略：返回默认值/抛自定义异常/等待
• 监控集成：通过@AfterThrowing记录限流事件
• 动态配置：结合配置中心实时更新限流阈值
• 注解继承：通过@Inherited支持类级别注解

5. 常见错误

• 线程阻塞：误用semaphore.acquire()导致线程堆积
• 内存泄漏：未清理不再使用的methodKey
• 时间精度问题：使用System.currentTimeMillis()而非nanoTime
• 分布式不一致：Redis操作未用Lua保证原子性

6. 底层原理扩展

• AOP代理机制：JDK动态代理与CGLIB字节码增强区别
• 执行链：ReflectiveMethodInvocation如何管理拦截器
• 注解解析：AnnotationUtils处理注解继承的算法
• 资源清理：@PreDestroy中关闭线程池