分布式限流器框架 eval-rate-limiter

分布式限流器框架 eval-rate-limiter

前言

基于 redis 实现的分布式限流器，实现如下效果

• 对每个应用节点进行限流
  • 有需要的可以根据实际业务定制 rateLimiterKey 的生成规则，目前是针对 ip+port 进行限流
• 控制限流的速率->次数/分钟

源码已上传 github

• https://github.com/huajiexiewenfeng/eval-rate-limiter

设计

流程图

• 时间窗口&最大条数可以配置

• 采用 redis 的过期时间+increment 来实现时间窗口和计数功能

• synchronized 防止多线程获取 count 产生一致性问题

• key 采用实例的 host+port 来实现各个实例控制自己的发送速率

核心方法

tryAcquire 获取通信证

• 获取当前的请求次数 currentCount
• 比较 currentCount 和 MaxCount 的大小
  • 小于，那么 count++，返回 true，表示获取通行证成功
  • 否则，返回 false，表示获取通信证失败

    /**
     * 尝试获取限流通行证
     *
     * @return true表示允许通过，false表示被限流
     */
    public synchronized boolean tryAcquire() {
        if (!properties.getEnable()) {
            return true;
        }
        try {
            int currentCount = getCurrentRequestCount();
            if (currentCount >= properties.getMaxCount()) {
                logger.warn("Rate limit exceeded - window: {} minutes, max: {}, current: {}",
                        properties.getWindowMinutes(),
                        properties.getMaxCount(),
                        currentCount);
                return false;
            }
            incrementRequestCount();
            return true;
        } catch (Exception e) {
            logger.error("Failed to acquire rate limit token for key: {}", rateLimiterKey, e);
            // 在限流器异常时默认放行，保证系统可用性
            return true;
        }
    }

增加访问次数 incrementRequestCount

• 使用 redis increment 计数+1
• 第一次设置过期时间为时间窗口

    private void incrementRequestCount() {
        Long count = valueOperations.increment(rateLimiterKey, 1);
        if (count != null && count == 1) {
            // 首次设置时初始化过期时间
            redisTemplate.expire(rateLimiterKey, properties.getWindowMinutes(), TimeUnit.MINUTES);
        }
    }

生成分布式 key generateRateLimiterKey

    private String generateRateLimiterKey() {
        String port = environment.getProperty("server.port", "unknown");
        String host = getLocalHostAddress();
        return String.format("%s:%s:%s", RATE_LIMITER_KEY_PREFIX, host, port);
    }

测试

测试代码

@SpringBootApplication
@EnableEvalRateLimiter
@RestController
@RequestMapping("/demo1")
public class Demo1Application {

    @Autowired
    private RateLimiter rateLimiter;

    public static void main(String[] args) {
        SpringApplication.run(Demo1Application.class, args);
    }

    @GetMapping("/test")
    public String test(@RequestParam(value = "count", defaultValue = "105") int count) {
        for (int i = 0; i < count; i++) {
            doTest(i);
        }
        return "hello";
    }

    private void doTest(int count) {
        System.out.println("当前请求次数：" + count + "，节点限流剩余次数：" + rateLimiter.getRemainingRequests());
        if (!rateLimiter.tryAcquire()) {
            // 5秒后再发送
            try {
                Thread.sleep(5000L);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            doTest(count);
        }
    }

}

application.properties 配置

spring.application.name=${APPLICATION_NAME:demo1}
server.port=${SERVER_PORT:8091}

#redis
eval.rate.limiter.redis.database=${REDIS_DB_INDEX:2}
eval.rate.limiter.redis.host=${REDIS_HOST:127.0.0.1}
eval.rate.limiter.redis.port=${REDIS_PORT:6379}
eval.rate.limiter.redis.password=${REDIS_INFRA_PASSWORD:1234567a}
#1分钟
eval.rate.limiter.redis.windowMinutes=${EVAL_RATE_LIMITER_WINDOW_MINUTES:1}
#100次
eval.rate.limiter.redis.maxCount=${EVAL_RATE_LIMITER_MAX_COUNT:100}

结果

到 1 分钟时间之后，又重新计数

Redis 客户端

• key：host+ip，限制每个机器+端口
• value：每访问一次，value+1，value 到达100（配置）后，不在增加