常见限流算法

GPT摘要

文章主要讨论了限流算法的几种实现方式及其特点。首先提到固定窗口限流算法（Fixed Window），通过设定时间窗口内的最大请求数来限流。该方法实现简便，但存在临界突发问题，即在窗口边界可能出现流量突增的情况（如两个窗口交界处可能允许两倍流量通过）。滑动窗口限流算法（Sliding Window）通过记录请求时间戳队列动态计算窗口内请求数，相比固定窗口更精确，但内存消耗大且清理过期数据可能影响性能。平滑滑动窗口（Smooth Sliding Window）进一步优化，将大窗口划分为多个子窗口，通过周期重置子窗口计数来平滑流量统计。使用锁保证线程安全，适合热点检测等场景。漏桶算法（Leaky Bucket）以恒定速率处理请求，实现流量整形。通过控制漏水速率和桶容量来平滑流量，但对突发流量处理较严格。文中代码示例仅实现容量检查，未完全体现匀速处理特性。令牌桶算法（Token Bucket）定期向桶添加令牌，允许突发流量消耗积累的令牌。通过控制令牌生成速率和桶容量实现灵活限流，适合需要容忍短期突发的场景。各算法对比：固定窗口实现简单但有临界问题；滑动窗口精度高但资源消耗大；漏桶强制流量匀速；令牌桶允许突发但控制速率。选择时需权衡精度、性能和业务需求。

固定窗口

临界突发问题：在窗口的边界，可能有大量流量

public class FixedWindowRateLimiter {

    // 每个窗口允许的最大请求数
    private final int maxRequests;
    // 窗口大小，单位为毫秒
    private final long windowSizeInMillis;
    // 记录当前窗口开始的时间
    private long windowStart;
    // 当前窗口内的请求计数
    private final AtomicInteger requestCount;

    public FixedWindowRateLimiter(int maxRequests, long windowSizeInMillis) {
        this.maxRequests = maxRequests;
        this.windowSizeInMillis = windowSizeInMillis;
        this.windowStart = System.currentTimeMillis();
        this.requestCount = new AtomicInteger(0);
    }

    public synchronized boolean allowRequest() {
        long currentTime = System.currentTimeMillis();

        // 判断当前时间是否已经进入新的窗口
        if (currentTime - windowStart >= windowSizeInMillis) {
            // 重置窗口
            windowStart = currentTime;
            requestCount.set(0);
        }

        // 判断是否可以在当前窗口内处理请求
        if (requestCount.incrementAndGet() <= maxRequests) {
            return true; // 允许请求
        } else {
            return false; // 拒绝请求
        }
    }
}

滑动窗口

简单实现：内存占用可能较大，每次请求都需要清理过期数据，可能影响性能

import java.util.LinkedList;
import java.util.Queue;

public class SlidingWindowRateLimiter {
    private final long windowSizeInMillis;
    private final int maxRequests;
    private final Queue<Long> requestTimestamps;

    public SlidingWindowRateLimiter(int windowSizeInSeconds, int maxRequests) {
        this.windowSizeInMillis = windowSizeInSeconds * 1000L;
        this.maxRequests = maxRequests;
        this.requestTimestamps = new LinkedList<>();
    }

    public synchronized boolean allowRequest() {
        long currentTime = System.currentTimeMillis();
        
        // 移除窗口外的请求时间戳
        while (!requestTimestamps.isEmpty() && currentTime - requestTimestamps.peek() >= windowSizeInMillis) {
            requestTimestamps.poll();
        }

        // 检查是否超过限制
        if (requestTimestamps.size() < maxRequests) {
            requestTimestamps.offer(currentTime);
            return true;
        }

        return false;
    }

    public static void main(String[] args) throws InterruptedException {
        SlidingWindowRateLimiter rateLimiter = new SlidingWindowRateLimiter(1, 3); // 1秒内最多3个请求

        for (int i = 0; i < 5; i++) {
            System.out.println("Request " + (i + 1) + " allowed: " + rateLimiter.allowRequest());
            Thread.sleep(300); // 每次请求间隔300毫秒
        }
    }
}

平滑滑动窗口（划分多个小窗口）

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;

public class SmoothSlidingWindowRateLimiter {
    private final long windowSizeInMillis;
    private final int maxRequests;
    private final int subWindowCount;
    private final AtomicInteger[] subWindows;
    private final ReentrantLock lock = new ReentrantLock();
    private long currentWindowStart;

    public SmoothSlidingWindowRateLimiter(int windowSizeInSeconds, int maxRequests, int subWindowCount) {
        this.windowSizeInMillis = windowSizeInSeconds * 1000L;
        this.maxRequests = maxRequests;
        this.subWindowCount = subWindowCount;
        this.subWindows = new AtomicInteger[subWindowCount];
        for (int i = 0; i < subWindowCount; i++) {
            subWindows[i] = new AtomicInteger(0);
        }
        this.currentWindowStart = System.currentTimeMillis();
    }

    public boolean allowRequest() {
        long now = System.currentTimeMillis();
        lock.lock();
        try {
            if (now - currentWindowStart > windowSizeInMillis) {
                slideWindow(now);
            }
            
            long totalRequests = getTotalRequests();
            if (totalRequests < maxRequests) {
                int index = (int) (((now - currentWindowStart) * subWindowCount) / windowSizeInMillis);
                subWindows[index].incrementAndGet();
                return true;
            }
            return false;
        } finally {
            lock.unlock();
        }
    }

    private void slideWindow(long now) {
        long newWindowStart = now - windowSizeInMillis;
        long windowDiff = newWindowStart - currentWindowStart;
        int resetCount = (int) ((windowDiff * subWindowCount) / windowSizeInMillis);
        
        if (resetCount >= subWindowCount) {
            for (AtomicInteger subWindow : subWindows) {
                subWindow.set(0);
            }
        } else {
            for (int i = 0; i < resetCount; i++) {
                int index = (int) (((currentWindowStart - newWindowStart) * subWindowCount) / windowSizeInMillis + i) % subWindowCount;
                subWindows[index].set(0);
            }
        }
        currentWindowStart = newWindowStart;
    }

    private long getTotalRequests() {
        long total = 0;
        for (AtomicInteger subWindow : subWindows) {
            total += subWindow.get();
        }
        return total;
    }

    public static void main(String[] args) throws InterruptedException {
        SmoothSlidingWindowRateLimiter rateLimiter = new SmoothSlidingWindowRateLimiter(1, 10, 10); // 1秒内最多10个请求，分成10个小窗口

        for (int i = 0; i < 15; i++) {
            System.out.println("Request " + (i + 1) + " allowed: " + rateLimiter.allowRequest());
            Thread.sleep(100); // 每次请求间隔100毫秒
        }
    }
}

也常用于热点key的检测

漏桶算法

固定速率处理请求，处理任意流量更加平滑，可以实现流量整形；

public class LeakyBucket {
    private int capacity; //漏桶容量
    private int rate; //漏水速率
    private int water; //当前水量
    private Instant timestamp; //上次漏水时间

    public LeakyBucket(int capacity, int rate) {
        this.capacity = capacity;
        this.rate = rate;
        this.water = 0;
        this.timestamp = Instant.now();
    }

    public synchronized boolean allow() { //判断是否允许通过
        Instant now = Instant.now();
        long duration = now.toEpochMilli() - timestamp.toEpochMilli(); //计算距上次漏水过去了多久
        int outflow = (int) (duration * rate / 1000); //计算过去的时间内漏出的水量
        water = Math.max(0, water - outflow); //更新当前水量，不能小于0
        if (water < capacity) { //如果漏桶还没满，放行
            water++;
            timestamp = now;
            return true;
        }
        return false; //否则拒绝通过
    }
}

// 感觉这个代码只做到了  对超过桶容量的请求拒绝对接受的请求，好像并没有按照漏桶的速度进行处理，而是小于capacity就立马处理了？

令牌桶算法

通过控制桶中的令牌实现限流，可以处理一定的突发流量，比如处理一次洪峰。

public class TokenBucket {
    
    private long lastTime;  // 上次请求时间
    private double rate;    // 令牌放入速率
    private long capacity;  // 令牌桶容量
    private long tokens;    // 当前令牌数量
    
    public TokenBucket(double rate, long capacity) {
        this.lastTime = System.currentTimeMillis();
        this.rate = rate;
        this.capacity = capacity;
        this.tokens = capacity;
    }
    
    public synchronized boolean getToken() {
        long now = System.currentTimeMillis();
        long timeElapsed = now - lastTime;
        tokens += timeElapsed * rate;
        if (tokens > capacity) {
            tokens = capacity;
        }
        lastTime = now;
        if (tokens >= 1) {
            tokens--;
            return true;
        } else {
            return false;
        }
    }
}

参考

开发 > 高并发

#高并发 #限流

hotKey检测上一篇

3.kafka-rocketmq对比-MQ 下一篇