基于Redis分布式的限流

以下是基于 Redis 实现分布式限流的 Java 解决方案，包含多种限流算法和完整实现代码：

一、限流算法选择与实现

1. 固定窗口算法（Simple Rate Limiter）

public class RedisFixedWindowRateLimiter {
    private final StringRedisTemplate redisTemplate;
    private final String script = 
        "local current = redis.call('INCR', KEYS[1])\n" +
        "if current == 1 then\n" +
        "  redis.call('EXPIRE', KEYS[1], ARGV[1])\n" +
        "end\n" +
        "return current > tonumber(ARGV[2])";

    public RedisFixedWindowRateLimiter(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    /**
     * 固定窗口限流
     * @param key 限流标识
     * @param period 时间窗口（秒）
     * @param limit 最大请求数
     * @return 是否允许请求
     */
    public boolean isAllowed(String key, int period, int limit) {
        key = "rate_limit:" + key;
        return !redisTemplate.execute(
            new DefaultRedisScript<>(script, Boolean.class),
            Collections.singletonList(key),
            String.valueOf(period),
            String.valueOf(limit)
        );
    }
}

2. 滑动窗口算法（基于ZSet）

public class RedisSlidingWindowRateLimiter {
    private final StringRedisTemplate redisTemplate;
    private final RedisScript<Long> script;

    public RedisSlidingWindowRateLimiter(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.script = new DefaultRedisScript<>(
            "local key = KEYS[1]\n" +
            "local now = tonumber(ARGV[1])\n" +
            "local window = tonumber(ARGV[2])\n" +
            "local limit = tonumber(ARGV[3])\n" +
            "local start = now - window\n" +
            // 移除窗口外的记录
            "redis.call('ZREMRANGEBYSCORE', key, 0, start)\n" +
            // 统计窗口内请求数
            "local count = redis.call('ZCARD', key)\n" +
            // 判断是否超出限制
            "local allowed = count < limit\n" +
            // 若允许则添加当前请求时间戳
            "if allowed then redis.call('ZADD', key, now, now) end\n" +
            // 设置过期时间，避免冷key永久存在
            "redis.call('EXPIRE', key, window)\n" +
            "return allowed and 1 or 0",
            Long.class
        );
    }

    /**
     * 滑动窗口限流
     * @param key 限流标识
     * @param window 窗口大小（毫秒）
     * @param limit 窗口内最大请求数
     * @return 是否允许请求
     */
    public boolean isAllowed(String key, long window, int limit) {
        key = "sliding_window:" + key;
        long now = System.currentTimeMillis();
        return redisTemplate.execute(
            script, 
            Collections.singletonList(key),
            String.valueOf(now),
            String.valueOf(window),
            String.valueOf(limit)
        ) == 1L;
    }
}

3. 令牌桶算法（Token Bucket）

public class RedisTokenBucketRateLimiter {
    private final StringRedisTemplate redisTemplate;
    private final RedisScript<List<Long>> script;

    public RedisTokenBucketRateLimiter(StringRedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
        this.script = new DefaultRedisScript<>(
            "local tokens_key = KEYS[1]\n" +
            "local timestamp_key = KEYS[2]\n" +
            "local rate = tonumber(ARGV[1])\n" +  // 令牌生成速率（个/秒）
            "local capacity = tonumber(ARGV[2])\n" +  // 桶容量
            "local now = tonumber(ARGV[3])\n" +  // 当前时间戳（毫秒）
            "local requested = tonumber(ARGV[4])\n" +  // 请求的令牌数
            "local last_tokens = tonumber(redis.call('GET', tokens_key) or capacity)\n" +
            "local last_refreshed = tonumber(redis.call('GET', timestamp_key) or 0)\n" +
            "local delta = math.max(0, now - last_refreshed)\n" +  // 距离上次刷新的时间差
            "local filled_tokens = math.min(capacity, last_tokens + (delta * rate / 1000))\n" +  // 计算当前令牌数
            "local allowed = filled_tokens >= requested\n" +  // 是否允许请求
            "local new_tokens = filled_tokens\n" +
            "if allowed then\n" +
            "  new_tokens = filled_tokens - requested\n" +
            "end\n" +
            "redis.call('SET', tokens_key, new_tokens, 'EX', 3600)\n" +  // 设置1小时过期
            "redis.call('SET', timestamp_key, now, 'EX', 3600)\n" +
            "return { allowed and 1 or 0, new_tokens }",
            List.class
        );
    }

    /**
     * 令牌桶限流
     * @param key 限流标识
     * @param rate 令牌生成速率（个/秒）
     * @param capacity 桶容量
     * @param requested 请求的令牌数（通常为1）
     * @return 是否允许请求
     */
    public boolean isAllowed(String key, int rate, int capacity, int requested) {
        key = "token_bucket:" + key;
        String tokensKey = key + ":tokens";
        String timestampKey = key + ":timestamp";
        long now = System.currentTimeMillis();
        
        List<Long> result = redisTemplate.execute(
            script, 
            Arrays.asList(tokensKey, timestampKey),
            String.valueOf(rate),
            String.valueOf(capacity),
            String.valueOf(now),
            String.valueOf(requested)
        );
        
        return result != null && result.get(0) == 1L;
    }
}

二、注解式限流实现（基于Spring AOP）

1. 自定义限流注解

@Target({ElementType.METHOD, ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface RateLimit {
    String key() default "rate_limit";
    int period() default 60;  // 时间窗口（秒）
    int limit() default 100;  // 最大请求数
    RateLimitType type() default RateLimitType.COMMON;  // 限流类型
}

public enum RateLimitType {
    COMMON,  // 普通限流
    IP,      // 基于IP限流
    USER     // 基于用户限流
}

2. AOP切面实现

@Aspect
@Component
public class RateLimitAspect {
    private final RedisFixedWindowRateLimiter rateLimiter;

    public RateLimitAspect(StringRedisTemplate redisTemplate) {
        this.rateLimiter = new RedisFixedWindowRateLimiter(redisTemplate);
    }

    @Around("@annotation(com.example.ratelimit.RateLimit)")
    public Object around(ProceedingJoinPoint joinPoint) throws Throwable {
        MethodSignature signature = (MethodSignature) joinPoint.getSignature();
        Method method = signature.getMethod();
        RateLimit rateLimit = method.getAnnotation(RateLimit.class);
        
        String key = getKey(rateLimit, joinPoint);
        boolean allowed = rateLimiter.isAllowed(
            key, 
            rateLimit.period(), 
            rateLimit.limit()
        );
        
        if (!allowed) {
            throw new RuntimeException("请求过于频繁，请稍后再试");
        }
        
        return joinPoint.proceed();
    }

    private String getKey(RateLimit rateLimit, ProceedingJoinPoint joinPoint) {
        switch (rateLimit.type()) {
            case IP:
                return getClientIp() + "_" + rateLimit.key();
            case USER:
                return getUserId() + "_" + rateLimit.key();
            default:
                return methodSignatureToKey(joinPoint);
        }
    }

    // 获取客户端IP（实际实现需从Request中获取）
    private String getClientIp() {
        // 实现略
        return "127.0.0.1";
    }

    // 获取用户ID（实际实现需从SecurityContext或Session中获取）
    private String getUserId() {
        // 实现略
        return "user_123";
    }

    private String methodSignatureToKey(ProceedingJoinPoint joinPoint) {
        MethodSignature signature = (MethodSignature) joinPoint.getSignature();
        return signature.getDeclaringTypeName() + "." + signature.getName();
    }
}

三、限流熔断与降级策略

1. 熔断降级处理器

@Component
public class RateLimitFallbackHandler {
    /**
     * 默认降级方法（可自定义返回值）
     */
    public Object defaultFallback(JoinPoint joinPoint, Throwable ex) {
        return Result.fail("系统繁忙，请稍后再试", 503);
    }
}

2. 整合Sentinel实现熔断降级

@Service
public class UserService {
    @RateLimit(key = "getUser", limit = 200)
    @SentinelResource(value = "getUser", fallback = "getUserFallback")
    public User getUser(Long userId) {
        // 业务逻辑
        return userDao.getUserById(userId);
    }

    public User getUserFallback(Long userId, Throwable ex) {
        // 限流或熔断后的降级逻辑
        return new User(-1L, "系统繁忙，请稍后再试");
    }
}

四、配置与使用示例

1. Redis配置

@Configuration
public class RedisConfig {
    @Bean
    public StringRedisTemplate stringRedisTemplate(RedisConnectionFactory factory) {
        StringRedisTemplate template = new StringRedisTemplate();
        template.setConnectionFactory(factory);
        template.setKeySerializer(new StringRedisSerializer());
        template.setValueSerializer(new StringRedisSerializer());
        return template;
    }
}

2. 使用注解限流

@RestController
@RequestMapping("/api")
public class ApiController {
    @GetMapping("/data")
    @RateLimit(key = "getData", period = 60, limit = 100)
    public ResponseData getData() {
        // 业务逻辑
        return ResponseData.success();
    }
}

五、高级特性与优化建议

1. 限流指标监控

@Component
public class RateLimitMetrics {
    private final StringRedisTemplate redisTemplate;
    private final MeterRegistry meterRegistry;

    public RateLimitMetrics(StringRedisTemplate redisTemplate, MeterRegistry meterRegistry) {
        this.redisTemplate = redisTemplate;
        this.meterRegistry = meterRegistry;
    }

    /**
     * 收集限流指标
     */
    @Scheduled(fixedDelay = 60_000)
    public void collectMetrics() {
        Set<String> keys = redisTemplate.keys("rate_limit:*");
        if (keys != null) {
            for (String key : keys) {
                String count = redisTemplate.opsForValue().get(key);
                if (count != null) {
                    String metricName = key.replace("rate_limit:", "ratelimit.");
                    meterRegistry.gauge(metricName, Long.parseLong(count));
                }
            }
        }
    }
}

2. 动态限流规则

@Service
public class DynamicRateLimitService {
    private final StringRedisTemplate redisTemplate;
    private final Map<String, RateLimitConfig> configMap = new ConcurrentHashMap<>();

    @PostConstruct
    public void init() {
        // 从配置中心加载规则
        loadRulesFromConfigCenter();
        // 注册配置变更监听器
        registerConfigChangeListener();
    }

    public boolean isAllowed(String key) {
        RateLimitConfig config = configMap.get(key);
        if (config == null) {
            return true;
        }
        return new RedisFixedWindowRateLimiter(redisTemplate)
            .isAllowed(key, config.getPeriod(), config.getLimit());
    }
}

六、不同限流算法对比与选择

七、注意事项与最佳实践

1. 性能优化：

   • 使用 Lua 脚本保证原子性
   • 批量操作减少网络开销
   • 合理设置过期时间避免冷key占用内存

2. 异常处理：

   • Redis连接异常时的降级策略（如允许请求或拒绝所有请求）
   • 熔断机制防止级联故障

3. 监控与告警：

   • 监控限流命中率、QPS、拒绝率等指标
   • 设置阈值告警（如拒绝率超过50%时触发）

4. 分布式一致性：

   • 多节点部署时确保时间同步（NTP服务）
   • 优先选择Redis Cluster或哨兵模式保证高可用

通过以上实现，你可以基于 Redis 构建高性能、高可用的分布式限流系统，有效保护后端服务免受流量冲击。