Java滑动窗口限流算法原理解析

时间轴：|-------第1秒-------|-------第2秒-------|
请求：   [95个]           [95个]
              ^                ^
           最后10ms         开头10ms

当前时间：1.35秒
窗口范围：[0.35s ~ 1.35s]（共1秒）
格子划分：
| 0.35-0.45 | 0.45-0.55 | 0.55-0.65 | 0.65-0.75 | 0.75-0.85 | 0.85-0.95 | 0.95-1.05 | 1.05-1.15 | 1.15-1.25 | 1.25-1.35 |
|    8      |    12     |    5      |    20     |    15     |    10     |    25     |    18     |    7      |    10     |
总请求数 = 8+12+5+20+15+10+25+18+7+10 = 130
如果阈值是100，当前请求就会被拒绝。

T1时刻窗口：[0.0 ~ 1.0]  包含格子 G1,G2,G3,G4,G5,G6,G7,G8,G9,G10
T2时刻（过了100ms）：
- 丢弃 G1（已经滑出窗口）
- 新增 G11（进入窗口）
- 新窗口：[0.1 ~ 1.1] 包含 G2~G11

import java.util.Arrays;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
/**
 * 基于环形数组的滑动窗口限流器
 * 
 * 核心设计：
 * 1. 用环形数组存储每个格子的请求计数
 * 2. 通过时间戳计算当前请求落在哪个格子
 * 3. 滑动时清空过期格子的计数
 */
public class SlidingWindowRateLimiter {
    /** 窗口大小（毫秒） */
    private final long windowSizeMs;
    /** 格子数量 */
    private final int gridCount;
    /** 每个格子的时长（毫秒） */
    private final long gridDurationMs;
    /** 限流阈值 */
    private final int threshold;
    /** 环形数组，存储每个格子的请求数 */
    private final AtomicInteger[] grids;
    /** 每个格子最后更新的时间戳 */
    private final long[] gridTimestamps;
    /** 保护数组更新的锁 */
    private final ReentrantLock lock = new ReentrantLock();
    public SlidingWindowRateLimiter(long windowSizeMs, int gridCount, int threshold) {
        this.windowSizeMs = windowSizeMs;
        this.gridCount = gridCount;
        this.gridDurationMs = windowSizeMs / gridCount;
        this.threshold = threshold;
        this.grids = new AtomicInteger[gridCount];
        this.gridTimestamps = new long[gridCount];
        for (int i = 0; i < gridCount; i++) {
            grids[i] = new AtomicInteger(0);
            gridTimestamps[i] = -1;
        }
    }
    /**
     * 尝试获取一个请求配额
     * @return true-允许通过，false-被限流
     */
    public boolean tryAcquire() {
        long now = System.currentTimeMillis();
        // 计算当前时间落在哪个格子
        int currentGridIndex = (int) ((now / gridDurationMs) % gridCount);
        lock.lock();
        try {
            // 检查当前格子是否过期（不是本次窗口的）
            long gridStartTime = (now / gridDurationMs) * gridDurationMs;
            if (gridTimestamps[currentGridIndex] != gridStartTime) {
                // 格子过期，重置计数
                grids[currentGridIndex].set(0);
                gridTimestamps[currentGridIndex] = gridStartTime;
            }
            // 计算当前窗口内的总请求数
            int total = calculateCurrentWindowCount(now);
            if (total < threshold) {
                grids[currentGridIndex].incrementAndGet();
                return true;
            }
            return false;
        } finally {
            lock.unlock();
        }
    }
    /**
     * 计算当前滑动窗口内的请求总数
     */
    private int calculateCurrentWindowCount(long now) {
        int total = 0;
        long windowStart = now - windowSizeMs;
        for (int i = 0; i < gridCount; i++) {
            if (gridTimestamps[i] > windowStart) {
                total += grids[i].get();
            }
        }
        return total;
    }
    /**
     * 获取当前窗口的实时QPS（用于监控）
     */
    public int getCurrentQps() {
        lock.lock();
        try {
            return calculateCurrentWindowCount(System.currentTimeMillis());
        } finally {
            lock.unlock();
        }
    }
    @Override
    public String toString() {
        return "SlidingWindowRateLimiter{" +
                "windowSizeMs=" + windowSizeMs +
                ", gridCount=" + gridCount +
                ", threshold=" + threshold +
                ", currentQps=" + getCurrentQps() +
                '}';
    }
    // ==================== 测试代码 ====================
    public static void main(String[] args) throws InterruptedException {
        // 1秒窗口，10个格子，阈值100
        SlidingWindowRateLimiter limiter = new SlidingWindowRateLimiter(1000, 10, 100);
        System.out.println("=== 测试1：正常请求 ===");
        int pass = 0, reject = 0;
        for (int i = 0; i < 120; i++) {
            if (limiter.tryAcquire()) {
                pass++;
            } else {
                reject++;
            }
        }
        System.out.println("通过: " + pass + ", 拒绝: " + reject);
        // 预期：通过100，拒绝20
        System.out.println("\n=== 测试2：等待窗口滑动后再次请求 ===");
        Thread.sleep(200); // 等待200ms
        pass = 0; reject = 0;
        for (int i = 0; i < 50; i++) {
            if (limiter.tryAcquire()) {
                pass++;
            } else {
                reject++;
            }
        }
        System.out.println("通过: " + pass + ", 拒绝: " + reject);
        System.out.println("\n=== 测试3：模拟突发流量 ===");
        SlidingWindowRateLimiter limiter2 = new SlidingWindowRateLimiter(1000, 10, 10);
        // 快速发送20个请求
        pass = 0; reject = 0;
        for (int i = 0; i < 20; i++) {
            if (limiter2.tryAcquire()) {
                pass++;
            } else {
                reject++;
            }
        }
        System.out.println("通过: " + pass + ", 拒绝: " + reject);
        // 预期：通过10，拒绝10
        System.out.println("\n=== 运行结果 ===");
        System.out.println("本地内存版滑动窗口限流器测试完成！");
    }
}

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.Pipeline;
import redis.clients.jedis.Response;
/**
 * 基于Redis ZSet的分布式滑动窗口限流器
 * 
 * 核心原理：
 * 1. 用ZSet的score存储请求时间戳
 * 2. 每次请求时，先删除窗口外的旧记录（ZREMRANGEBYSCORE）
 * 3. 统计当前窗口内的记录数（ZCARD）
 * 4. 如果未超限，添加当前请求记录（ZADD）
 * 
 * 踩坑提醒：这个实现需要Redis 2.6+支持lua脚本才能保证原子性！
 * 下面提供的是Pipeline版本，生产环境建议用lua脚本版本。
 */
public class RedisSlidingWindowRateLimiter {
    private final JedisPool jedisPool;
    private final String keyPrefix;
    private final long windowSizeMs;
    private final int threshold;
    public RedisSlidingWindowRateLimiter(String redisHost, int redisPort, 
                                          String keyPrefix, long windowSizeMs, int threshold) {
        JedisPoolConfig config = new JedisPoolConfig();
        config.setMaxTotal(100);
        config.setMaxIdle(20);
        this.jedisPool = new JedisPool(config, redisHost, redisPort);
        this.keyPrefix = keyPrefix;
        this.windowSizeMs = windowSizeMs;
        this.threshold = threshold;
    }
    /**
     * 尝试获取配额（Pipeline版本，非原子性，适合容忍少量误差的场景）
     */
    public boolean tryAcquire(String resource) {
        String key = keyPrefix + ":" + resource;
        long now = System.currentTimeMillis();
        long windowStart = now - windowSizeMs;
        try (Jedis jedis = jedisPool.getResource()) {
            Pipeline pipeline = jedis.pipelined();
            // 1. 删除窗口外的旧记录
            pipeline.zremrangeByScore(key, 0, windowStart);
            // 2. 统计当前窗口内的记录数
            Response<Long> countResponse = pipeline.zcard(key);
            // 3. 执行前两条命令
            pipeline.sync();
            long currentCount = countResponse.get();
            if (currentCount < threshold) {
                // 4. 添加当前请求记录
                jedis.zadd(key, now, now + "_" + Thread.currentThread().getId());
                // 5. 设置过期时间，防止key残留
                jedis.pexpire(key, windowSizeMs);
                return true;
            }
            return false;
        }
    }
    /**
     * 原子性版本：使用Lua脚本
     * 生产环境强烈推荐用这个！
     */
    public boolean tryAcquireAtomic(String resource) {
        String key = keyPrefix + ":" + resource;
        long now = System.currentTimeMillis();
        long windowStart = now - windowSizeMs;
        // Lua脚本保证整个操作的原子性
        String luaScript = 
            "local key = KEYS[1] " +
            "local now = tonumber(ARGV[1]) " +
            "local windowStart = tonumber(ARGV[2]) " +
            "local threshold = tonumber(ARGV[3]) " +
            "local expire = tonumber(ARGV[4]) " +
            "-- 删除窗口外的记录 " +
            "redis.call('ZREMRANGEBYSCORE', key, 0, windowStart) " +
            "-- 获取当前窗口内记录数 " +
            "local count = redis.call('ZCARD', key) " +
            "if count < threshold then " +
            "    redis.call('ZADD', key, now, ARGV[5]) " +
            "    redis.call('PEXPIRE', key, expire) " +
            "    return 1 " +
            "else " +
            "    return 0 " +
            "end";
        try (Jedis jedis = jedisPool.getResource()) {
            Object result = jedis.eval(luaScript, 
                Arrays.asList(key), 
                Arrays.asList(
                    String.valueOf(now),
                    String.valueOf(windowStart),
                    String.valueOf(threshold),
                    String.valueOf(windowSizeMs),
                    now + "_" + Thread.currentThread().getId() + "_" + Math.random()
                )
            );
            return Long.valueOf(1).equals(result);
        }
    }
    /**
     * 获取当前QPS（用于监控）
     */
    public long getCurrentQps(String resource) {
        String key = keyPrefix + ":" + resource;
        long windowStart = System.currentTimeMillis() - windowSizeMs;
        try (Jedis jedis = jedisPool.getResource()) {
            jedis.zremrangeByScore(key, 0, windowStart);
            return jedis.zcard(key);
        }
    }
    public void close() {
        if (jedisPool != null && !jedisPool.isClosed()) {
            jedisPool.close();
        }
    }
    // ==================== 测试代码 ====================
    public static void main(String[] args) {
        // 注意：需要本地启动Redis，端口6379
        RedisSlidingWindowRateLimiter limiter = new RedisSlidingWindowRateLimiter(
            "localhost", 6379, "rate_limit", 1000, 10
        );
        System.out.println("=== Redis分布式滑动窗口限流测试 ===");
        String resource = "api:/order/create";
        int pass = 0, reject = 0;
        for (int i = 0; i < 15; i++) {
            if (limiter.tryAcquireAtomic(resource)) {
                pass++;
                System.out.println("请求 " + (i+1) + " -> 通过");
            } else {
                reject++;
                System.out.println("请求 " + (i+1) + " -> 被拒绝");
            }
        }
        System.out.println("\n总计：通过=" + pass + ", 拒绝=" + reject);
        System.out.println("当前QPS：" + limiter.getCurrentQps(resource));
        limiter.close();
    }
}

// 简化的LeapArray核心结构
public abstract class LeapArray<T> {
    // 窗口时长（默认1秒）
    protected int windowLengthInMs;
    // 采样数量（默认2个，即每个窗口500ms）
    protected int sampleCount;
    // 窗口数组（环形）
    protected final AtomicReferenceArray<WindowWrap<T>> array;
    // 计算当前时间对应的窗口索引
    private int calculateTimeIdx(long timeMillis) {
        long timeId = timeMillis / windowLengthInMs;
        return (int) (timeId % array.length());
    }
    // 计算窗口开始时间
    protected long calculateWindowStart(long timeMillis) {
        return timeMillis - timeMillis % windowLengthInMs;
    }
}

// Sentinel中滑动窗口的使用示例
Entry entry = null;
try {
    entry = SphU.entry("orderCreate");
    // 业务逻辑
} catch (BlockException e) {
    // 被限流或熔断
    return "系统繁忙，请稍后再试";
} finally {
    if (entry != null) {
        entry.exit();
    }
}