Redisson实现限流器详解：从原理到实践

什么是限流器？

限流器（Rate Limiter）是一种控制请求频率的机制，用于保护系统免受过多请求的冲击。想象一下，你开了一家餐厅，如果同时涌入1000个客人，厨房肯定忙不过来，这时候就需要"限流"——控制进入餐厅的人数。

限流的常见场景

// 场景1：API接口限流
@RestController
public class UserController {
    @GetMapping("/api/user/{id}")
    public User getUser(@PathVariable Long id) {
        // 每个用户每分钟最多请求100次
        // 如果超过限制，返回429状态码
    }
}

// 场景2：短信发送限流
public class SmsService {
    public void sendSms(String phone, String content) {
        // 每个手机号每分钟最多发送3条短信
        // 防止短信轰炸
    }
}

// 场景3：登录限流
public class LoginService {
    public boolean login(String username, String password) {
        // 每个IP每分钟最多尝试5次登录
        // 防止暴力破解
    }
}

限流算法介绍

1. 固定窗口算法

算法原理

固定窗口算法将时间划分为固定大小的窗口（如1分钟），在每个窗口内统计请求次数。当请求次数超过限制时，拒绝后续请求。

public class FixedWindowRateLimiter {
    // 限制次数：每个窗口最多允许的请求数
    private final int limit;
    // 窗口大小：时间窗口的毫秒数
    private final long windowSize;
    // 存储每个key对应的窗口信息
    private final Map<String, Window> windows = new ConcurrentHashMap<>();
    
    public FixedWindowRateLimiter(int limit, long windowSize) {
        this.limit = limit;
        this.windowSize = windowSize;
    }
    
    public boolean tryAcquire(String key) {
        long currentTime = System.currentTimeMillis();
        // 计算当前时间所在的窗口开始时间
        // 例如：当前时间1234567890，窗口大小60000ms
        // 窗口开始时间 = 1234567890 - (1234567890 % 60000) = 1234560000
        long windowStart = currentTime - (currentTime % windowSize);
        
        // 获取或创建窗口对象
        Window window = windows.computeIfAbsent(key, k -> new Window(windowStart));
        
        // 检查窗口是否过期
        if (currentTime >= window.startTime + windowSize) {
            // 窗口过期，重置窗口
            window.reset(windowStart);
        }
        
        // 检查是否超过限制
        if (window.count < limit) {
            window.count++;
            return true;
        }
        
        return false;
    }
    
    // 窗口内部类，存储窗口的统计信息
    private static class Window {
        long startTime;  // 窗口开始时间
        int count;      // 当前窗口内的请求计数
        
        Window(long startTime) {
            this.startTime = startTime;
            this.count = 0;
        }
        
        // 重置窗口
        void reset(long newStartTime) {
            this.startTime = newStartTime;
            this.count = 0;
        }
    }
}

固定窗口算法的特点

优点：

1. 实现简单：逻辑清晰，容易理解和实现

1. 内存占用少：只需要存储窗口开始时间和计数

1. 性能好：计算量小，响应速度快

缺点：

1. 窗口边界效应：在窗口切换时可能出现流量突增

1. 限流不均匀：无法平滑处理请求分布

2. 滑动窗口算法

算法原理

滑动窗口算法通过维护一个请求时间队列，动态计算当前时间窗口内的请求数量。窗口是"滑动"的，能够更精确地控制请求频率。

public class SlidingWindowRateLimiter {
    // 限制次数：时间窗口内最多允许的请求数
    private final int limit;
    // 窗口大小：时间窗口的毫秒数
    private final long windowSize;
    // 存储每个key的请求时间队列
    private final Map<String, Queue<Long>> requests = new ConcurrentHashMap<>();
    
    public SlidingWindowRateLimiter(int limit, long windowSize) {
        this.limit = limit;
        this.windowSize = windowSize;
    }
    
    public boolean tryAcquire(String key) {
        long currentTime = System.currentTimeMillis();
        // 获取或创建请求时间队列
        Queue<Long> requestTimes = requests.computeIfAbsent(key, k -> new LinkedList<>());
        
        // 清理过期的请求记录
        // 移除窗口外的请求时间
        while (!requestTimes.isEmpty() && 
               currentTime - requestTimes.peek() > windowSize) {
            requestTimes.poll();
        }
        
        // 检查当前窗口内的请求数量是否超过限制
        if (requestTimes.size() < limit) {
            // 添加当前请求时间
            requestTimes.offer(currentTime);
            return true;
        }
        
        return false;
    }
}

滑动窗口算法的特点

优点：

1. 限流均匀：没有窗口边界效应，限流更平滑

1. 精确控制：能够精确控制任意时间窗口内的请求数量

1. 实时性好：能够实时反映当前的请求状态

缺点：

1. 内存占用大：需要存储所有请求时间

1. 计算复杂度高：每次请求都需要清理过期数据

1. 性能相对较低：相比固定窗口，计算开销更大

3. 令牌桶算法

算法原理

令牌桶算法维护一个固定容量的令牌桶，以恒定速率向桶中放入令牌。请求需要获取令牌才能被处理，如果桶中没有令牌，则请求被拒绝。

public class TokenBucketRateLimiter {
    // 桶容量：桶中最多能存放的令牌数
    private final int capacity;
    // 令牌填充速率：每秒填充的令牌数
    private final double refillRate;
    // 存储每个key对应的令牌桶
    private final Map<String, Bucket> buckets = new ConcurrentHashMap<>();
    
    public TokenBucketRateLimiter(int capacity, double refillRate) {
        this.capacity = capacity;
        this.refillRate = refillRate;
    }
    
    public boolean tryAcquire(String key) {
        long currentTime = System.currentTimeMillis();
        // 获取或创建令牌桶
        Bucket bucket = buckets.computeIfAbsent(key, k -> new Bucket(capacity, currentTime));
        
        // 计算从上次填充到现在需要填充的令牌数
        long timePassed = currentTime - bucket.lastRefillTime;
        // 令牌数 = 时间间隔 * 填充速率 / 1000（转换为秒）
        double tokensToAdd = timePassed * refillRate / 1000.0;
        
        // 更新令牌数量，不能超过桶容量
        bucket.tokens = Math.min(capacity, bucket.tokens + tokensToAdd);
        bucket.lastRefillTime = currentTime;
        
        // 尝试获取一个令牌
        if (bucket.tokens >= 1) {
            bucket.tokens--;
            return true;
        }
        
        return false;
    }
    
    // 令牌桶内部类
    private static class Bucket {
        double tokens;           // 当前令牌数量
        long lastRefillTime;     // 上次填充时间
        
        Bucket(int capacity, long currentTime) {
            this.tokens = capacity;  // 初始化时桶是满的
            this.lastRefillTime = currentTime;
        }
    }
}

令牌桶算法的特点

优点：

1. 支持突发流量：桶满时可以处理突发请求

1. 限流平滑：令牌以恒定速率填充，限流更平滑

1. 灵活性高：可以调整桶容量和填充速率

1. 资源利用率高：能够充分利用系统资源

缺点：

1. 实现复杂：需要维护令牌填充逻辑

1. 内存占用：需要存储令牌数量和上次填充时间

1. 参数调优困难：需要合理设置容量和填充速率

Redisson限流器实现

1. Maven依赖

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson</artifactId>
    <version>3.24.3</version>
</dependency>

2. Redisson配置

import org.redisson.Redisson;
import org.redisson.api.RedissonClient;
import org.redisson.config.Config;

public class RedissonConfig {
    
    public static RedissonClient createRedissonClient() {
        Config config = new Config();
        
        // 单机模式
        config.useSingleServer()
                .setAddress("redis://localhost:6379")
                .setDatabase(0)
                .setPassword("your_password")  // 如果有密码
                .setConnectionPoolSize(64)
                .setConnectionMinimumIdleSize(10);
        
        // 集群模式（可选）
        /*
        config.useClusterServers()
                .addNodeAddress("redis://192.168.1.100:6379")
                .addNodeAddress("redis://192.168.1.101:6379")
                .addNodeAddress("redis://192.168.1.102:6379");
        */
        
        return Redisson.create(config);
    }
}

3. 基础限流器使用

import org.redisson.api.RRateLimiter;
import org.redisson.api.RateIntervalUnit;
import org.redisson.api.RateType;

public class RedissonRateLimiterExample {
    
    private final RedissonClient redissonClient;
    
    public RedissonRateLimiterExample() {
        this.redissonClient = RedissonConfig.createRedissonClient();
    }
    
    public void basicRateLimiter() {
        // 获取限流器
        RRateLimiter rateLimiter = redissonClient.getRateLimiter("myRateLimiter");
        
        // 初始化限流器
        // 参数说明：
        // RateType.OVERALL: 全局限流
        // RateType.PER_CLIENT: 每个客户端限流
        // 10: 速率（每秒10个请求）
        // 1: 时间单位（秒）
        // RateIntervalUnit.SECONDS: 时间单位类型
        rateLimiter.trySetRate(RateType.OVERALL, 10, 1, RateIntervalUnit.SECONDS);
        
        // 尝试获取许可
        boolean acquired = rateLimiter.tryAcquire(1);
        if (acquired) {
            System.out.println("请求被允许");
            // 执行业务逻辑
        } else {
            System.out.println("请求被限流");
            // 返回限流响应
        }
    }
    
    public void perUserRateLimiter(String userId) {
        // 为每个用户创建独立的限流器
        RRateLimiter userLimiter = redissonClient.getRateLimiter("user:" + userId + ":limiter");
        
        // 设置每个用户每分钟最多100次请求
        userLimiter.trySetRate(RateType.OVERALL, 100, 1, RateIntervalUnit.MINUTES);
        
        boolean acquired = userLimiter.tryAcquire(1);
        if (acquired) {
            System.out.println("用户 " + userId + " 的请求被允许");
        } else {
            System.out.println("用户 " + userId + " 的请求被限流");
        }
    }
}

4. 高级限流器功能

public class AdvancedRedissonRateLimiter {
    
    private final RedissonClient redissonClient;
    
    public AdvancedRedissonRateLimiter() {
        this.redissonClient = RedissonConfig.createRedissonClient();
    }
    
    // 带等待时间的限流器
    public void rateLimiterWithWait() {
        RRateLimiter rateLimiter = redissonClient.getRateLimiter("waitLimiter");
        rateLimiter.trySetRate(RateType.OVERALL, 5, 1, RateIntervalUnit.SECONDS);
        
        try {
            // 等待最多2秒获取许可
            boolean acquired = rateLimiter.tryAcquire(1, 2, TimeUnit.SECONDS);
            if (acquired) {
                System.out.println("成功获取许可");
            } else {
                System.out.println("等待超时，请求被限流");
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.out.println("等待被中断");
        }
    }
    
    // 批量获取许可
    public void batchAcquire() {
        RRateLimiter rateLimiter = redissonClient.getRateLimiter("batchLimiter");
        rateLimiter.trySetRate(RateType.OVERALL, 100, 1, RateIntervalUnit.MINUTES);
        
        // 尝试一次性获取10个许可
        boolean acquired = rateLimiter.tryAcquire(10);
        if (acquired) {
            System.out.println("成功获取10个许可");
        } else {
            System.out.println("许可不足，无法获取10个许可");
        }
    }
    
    // 获取限流器状态
    public void getLimiterStatus() {
        RRateLimiter rateLimiter = redissonClient.getRateLimiter("statusLimiter");
        rateLimiter.trySetRate(RateType.OVERALL, 10, 1, RateIntervalUnit.SECONDS);
        
        // 获取当前可用许可数
        long availablePermits = rateLimiter.availablePermits();
        System.out.println("当前可用许可数: " + availablePermits);
        
        // 获取等待队列长度
        long waitingThreads = rateLimiter.waitingThreads();
        System.out.println("等待线程数: " + waitingThreads);
    }
}

实际应用场景

1. Spring Boot集成

import org.springframework.web.bind.annotation.*;
import org.springframework.http.ResponseEntity;
import org.redisson.api.RRateLimiter;
import org.redisson.api.RateIntervalUnit;
import org.redisson.api.RateType;

@RestController
@RequestMapping("/api")
public class ApiController {
    
    private final RedissonClient redissonClient;
    
    public ApiController(RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
    }
    
    @GetMapping("/user/{id}")
    public ResponseEntity<?> getUser(@PathVariable Long id, 
                                   @RequestHeader("X-User-Id") String userId) {
        // 创建用户限流器
        RRateLimiter userLimiter = redissonClient.getRateLimiter("user:" + userId + ":api");
        userLimiter.trySetRate(RateType.OVERALL, 100, 1, RateIntervalUnit.MINUTES);
        
        // 尝试获取许可
        if (!userLimiter.tryAcquire(1)) {
            return ResponseEntity.status(429)
                    .body("请求过于频繁，请稍后再试");
        }
        
        // 执行业务逻辑
        User user = userService.getUser(id);
        return ResponseEntity.ok(user);
    }
    
    @PostMapping("/sms/send")
    public ResponseEntity<?> sendSms(@RequestBody SmsRequest request) {
        // 手机号限流
        RRateLimiter phoneLimiter = redissonClient.getRateLimiter("phone:" + request.getPhone());
        phoneLimiter.trySetRate(RateType.OVERALL, 3, 1, RateIntervalUnit.MINUTES);
        
        if (!phoneLimiter.tryAcquire(1)) {
            return ResponseEntity.status(429)
                    .body("该手机号发送短信过于频繁");
        }
        
        // 发送短信
        smsService.sendSms(request.getPhone(), request.getContent());
        return ResponseEntity.ok("短信发送成功");
    }
}

2. 自定义注解实现

import java.lang.annotation.ElementType;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface RateLimit {
    String key() default "";           // 限流键
    int limit() default 100;           // 限制次数
    int time() default 60;             // 时间窗口（秒）
    String message() default "请求过于频繁"; // 限流消息
}

// AOP实现
@Component
@Aspect
public class RateLimitAspect {
    
    private final RedissonClient redissonClient;
    
    public RateLimitAspect(RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
    }
    
    @Around("@annotation(rateLimit)")
    public Object around(ProceedingJoinPoint point, RateLimit rateLimit) throws Throwable {
        // 生成限流键
        String key = generateKey(point, rateLimit);
        
        // 获取限流器
        RRateLimiter limiter = redissonClient.getRateLimiter(key);
        limiter.trySetRate(RateType.OVERALL, rateLimit.limit(), rateLimit.time(), RateIntervalUnit.SECONDS);
        
        // 尝试获取许可
        if (!limiter.tryAcquire(1)) {
            throw new RuntimeException(rateLimit.message());
        }
        
        // 执行业务方法
        return point.proceed();
    }
    
    private String generateKey(ProceedingJoinPoint point, RateLimit rateLimit) {
        if (!rateLimit.key().isEmpty()) {
            return rateLimit.key();
        }
        
        // 自动生成键：类名:方法名
        String className = point.getTarget().getClass().getSimpleName();
        String methodName = point.getSignature().getName();
        return className + ":" + methodName;
    }
}

// 使用注解
@RestController
public class UserController {
    
    @RateLimit(limit = 10, time = 60, message = "登录尝试过于频繁")
    @PostMapping("/login")
    public ResponseEntity<?> login(@RequestBody LoginRequest request) {
        // 登录逻辑
        return ResponseEntity.ok("登录成功");
    }
    
    @RateLimit(key = "#request.phone", limit = 3, time = 60)
    @PostMapping("/verify")
    public ResponseEntity<?> verifyPhone(@RequestBody VerifyRequest request) {
        // 手机验证逻辑
        return ResponseEntity.ok("验证码发送成功");
    }
}

3. 分布式限流器

public class DistributedRateLimiter {
    
    private final RedissonClient redissonClient;
    
    public DistributedRateLimiter(RedissonClient redissonClient) {
        this.redissonClient = redissonClient;
    }
    
    // IP限流
    public boolean isIpAllowed(String ip, int limit, int timeWindow) {
        RRateLimiter ipLimiter = redissonClient.getRateLimiter("ip:" + ip);
        ipLimiter.trySetRate(RateType.OVERALL, limit, timeWindow, RateIntervalUnit.SECONDS);
        return ipLimiter.tryAcquire(1);
    }
    
    // 用户限流
    public boolean isUserAllowed(String userId, int limit, int timeWindow) {
        RRateLimiter userLimiter = redissonClient.getRateLimiter("user:" + userId);
        userLimiter.trySetRate(RateType.OVERALL, limit, timeWindow, RateIntervalUnit.SECONDS);
        return userLimiter.tryAcquire(1);
    }
    
    // 接口限流
    public boolean isApiAllowed(String apiKey, int limit, int timeWindow) {
        RRateLimiter apiLimiter = redissonClient.getRateLimiter("api:" + apiKey);
        apiLimiter.trySetRate(RateType.OVERALL, limit, timeWindow, RateIntervalUnit.SECONDS);
        return apiLimiter.tryAcquire(1);
    }
    
    // 全局限流
    public boolean isGlobalAllowed(int limit, int timeWindow) {
        RRateLimiter globalLimiter = redissonClient.getRateLimiter("global");
        globalLimiter.trySetRate(RateType.OVERALL, limit, timeWindow, RateIntervalUnit.SECONDS);
        return globalLimiter.tryAcquire(1);
    }
}