浅谈限流算法

福尔摩斯媛1个月前发布

110

限流

限流（Rate Limiting）是指在系统设计和开发过程中，对服务请求或数据传输速率进行控制的一种策略。

限流的目的是为了保护系统资源不被过度消耗，避免因短时间内大量请求导致的服务过载、响应延迟甚至崩溃等情况，从而保证系统的稳定性和可用性。

这篇文章我们就聊一下几种流行的限流算法。

固定窗口

固定窗口算法即固定时间窗口（单位时间）内限制请求的数量。

固定窗口算法算法将时间分成固定的窗口，通过将请求按照时间顺序放入时间窗口中，并计算该时间窗口内的请求数量，如果请求数量超出了限制，则拒绝该请求。

固定窗口算法优缺点

固定窗口实现简单，容易理解，方便上手。但是会明显出现时间窗口的 临界突变问题，即当请求的时间窗口与当前的时间窗口的临界点相交时可能导致限流异常。

如果我们每秒限制QPS为10，我们0.8秒的时候来了10个请求，1.2秒的时候又来了10个请求，1秒的时候窗口滑动了一次限流重置为0，这时候1.2秒来的请求将全部被处理。即0.8到1.2秒处理了20个请求，这明显超出了我们的预期。

滑动窗口

滑动窗口算法的核心思想是将单位时间周期分为n个小周期，分别记录每个小周期内接口的访问次数，并且根据时间滑动删除过期的小周期。

每个小周期都有自己独立的计数器，当请求到达时，将请求的时间戳映射到对应的小周期内，并将计数器加1。当小周期结束时，计数器清零。

假设单位时间还是1秒，滑动窗口算法把它划分为2个小周期，也就是滑动窗口（单位时间）被划分为2个小格子。每格表示0.5秒。每过0.5秒，时间窗口就会往右滑动一格。然后每个小周期，都有自己独立的计数器，如果请求是0.8秒到达的，0.5~ 1.0秒对应的计数器就会加1。

假设限流阀值还是10个请求，0.8秒时候来了10个请求，1.2秒的时候又来了10个请求。如果是固定窗口算法，是不会被限流的，但是滑动窗口的话，每过一个小周期，窗口会右移一个小格。过了1.0s这个点后，会右移一小格，1.2秒请求到来的时候，会计算0.5-1.0s和1.0到1.5秒两个小格的总请求数，这时候1.2s来的10个请求会被限流。

滑动窗口算法优缺点

精度高（可以通过调整时间窗口的大小来实现不同的限流效果）；可扩展性强（可以非常容易地与其他限流算法结合使用）。

算法相对复杂需要维护多个计数器；可能受到一些因素的影响无法完全解决临界突变问题（算法处理延迟、格子大小数量、请求突发等因素）。

漏桶

漏桶限流算法是一种常用的限流算法，它通过模拟一个漏桶来限制系统的流量。该算法将请求比作水，而漏桶则是用来控制水流量的工具。

漏桶限流算法的 核心思想 是模拟一个带有固定出水速率的漏桶来处理流入的请求或数据包：当有请求进来时，如果漏桶还有剩余容量，那么这个请求就可以成功放入漏桶中，如果漏桶已经满了，那么这个请求就被丢弃或者排队等待处理。

漏桶算法优缺点

漏桶限流算法的优点是简单易懂；能够平滑突发流量，为网络提供一个稳定的流量；精确严格的限流。

但该算法也有缺点，无法应对突发流量，漏桶的容量和出水速率可能会成为瓶颈；响应时间延长，如果漏桶累积过多可能导致请求等待较长时间才能被处理。

令牌桶

该算法通过维护一个令牌桶来实现流量控制，其中令牌桶中的令牌数量表示系统的可用流量，系统按照预设的速度持续不断的生成令牌放入令牌桶中。当有请求进来时，需要从令牌桶中获取一个令牌，如果没有令牌可用，则请求被拒绝。

令牌桶限流算法的 核心思想 是：在固定的速率往桶里加入令牌，桶的容量表示系统的最大流量。当有请求到来时，先检查桶中是否有可用的令牌，如果有则获取一个令牌并处理请求；如果没有可用的令牌，则根据桶中令牌的数量和请求的优先级等因素进行处理。

令牌桶算法优缺点

令牌桶限流算法的优点是简单易懂，能够灵活应对突发流量；精确控制请求处理的数量；可以与其他限流算法结合使用以实现更加灵活和高效的限流效果。

令牌桶算法实现相对复杂。

Sentinel中的算法

Sentinel中的限流算法我们主要来分析一下这几个类：

DefaultController
RateLimiterController
WarmUpController

DefaultController

DefaultController 对应的流控效果为快速失败。

`public class DefaultController implements TrafficShapingController {

    private static final int DEFAULT_AVG_USED_TOKENS = 0;

    private double count;

    private int grade;

public DefaultController(double count, int grade) {  
    this.count = count;  
    this.grade = grade;  
}

@Override  
public boolean canPass(Node node, int acquireCount) {  
    return canPass(node, acquireCount, false);  
}

@Override  
public boolean canPass(Node node, int acquireCount, boolean prioritized) {

            int curCount = avgUsedTokens(node);

            if (curCount + acquireCount > count) {  
        if (prioritized && grade == RuleConstant.FLOW_GRADE_QPS) {  
            long currentTime;  
            long waitInMs;  
            currentTime = TimeUtil.currentTimeMillis();  
            waitInMs = node.tryOccupyNext(currentTime, acquireCount, count);  
            if (waitInMs < OccupyTimeoutProperty.getOccupyTimeout()) {  
                node.addWaitingRequest(currentTime + waitInMs, acquireCount);  
                node.addOccupiedPass(acquireCount);  
                sleep(waitInMs);

                                    throw new PriorityWaitException(waitInMs);  
            }  
        }  
        return false;  
    }  
    return true;  
}

private int avgUsedTokens(Node node) {  
    if (node == null) {  
        return DEFAULT_AVG_USED_TOKENS;  
    }  
    return grade == RuleConstant.FLOW_GRADE_THREAD ? node.curThreadNum() : (int)(node.passQps());  
}

private void sleep(long timeMillis) {  
    try {  
        Thread.sleep(timeMillis);  
    } catch (InterruptedException e) {

                }  
}

RateLimiterController

RateLimiterController 对应的流控效果为排队等待，需要设置一个超时时间。

`public class RateLimiterController implements TrafficShapingController {

    private final int maxQueueingTimeMs;

    private final double count;

        private final AtomicLong latestPassedTime = new AtomicLong(-1);

public RateLimiterController(int timeOut, double count) {  
    this.maxQueueingTimeMs = timeOut;  
    this.count = count;  
}

@Override  
public boolean canPass(Node node, int acquireCount) {  
    return canPass(node, acquireCount, false);  
}

    @Override  
public boolean canPass(Node node, int acquireCount, boolean prioritized) {

            if (acquireCount <= 0) {  
        return true;  
    }

            if (count <= 0) {  
        return false;  
    }

    long currentTime = TimeUtil.currentTimeMillis();

            long costTime = Math.round(1.0 * (acquireCount) / count * 1000);

            long expectedTime = costTime + latestPassedTime.get();

    if (expectedTime <= currentTime) {

                    latestPassedTime.set(currentTime);  
        return true;  
    } else {

                    long waitTime = costTime + latestPassedTime.get() - TimeUtil.currentTimeMillis();  
        if (waitTime > maxQueueingTimeMs) {

                            return false;  
        } else {  
            long oldTime = latestPassedTime.addAndGet(costTime);  
            try {   

                                    waitTime = oldTime - TimeUtil.currentTimeMillis();  
                if (waitTime > maxQueueingTimeMs) {  
                    latestPassedTime.addAndGet(-costTime);  
                    return false;  
                }  
                if (waitTime > 0) {  
                    Thread.sleep(waitTime);  
                }  
                return true;  
            } catch (InterruptedException e) {  
            }  
        }  
    }  
    return false;  
}

可以看出，RateLimiterController 对应的是我们上文提到过的漏桶算法。

WarmUpController

WarmUpController 对应的流控效果就是 Warm Up。 WarmUpController 根据代码看的话相对复杂一些。

`public class WarmUpController implements TrafficShapingController {

    protected double count;

    private int coldFactor;

    protected int warningToken = 0;

    private int maxToken;

    protected double slope;

    protected AtomicLong storedTokens = new AtomicLong(0);

    protected AtomicLong lastFilledTime = new AtomicLong(0);

public WarmUpController(double count, int warmUpPeriodInSec, int coldFactor) {  
    construct(count, warmUpPeriodInSec, coldFactor);  
}

public WarmUpController(double count, int warmUpPeriodInSec) {  
    construct(count, warmUpPeriodInSec, 3);  
}

private void construct(double count, int warmUpPeriodInSec, int coldFactor) {

    if (coldFactor <= 1) {  
        throw new IllegalArgumentException("Cold factor should be larger than 1");  
    }

    this.count = count;

    this.coldFactor = coldFactor;

                    warningToken = (int)(warmUpPeriodInSec * count) / (coldFactor - 1);

                            maxToken = warningToken + (int)(2 * warmUpPeriodInSec * count / (1.0 + coldFactor));

                                    slope = (coldFactor - 1.0) / count / (maxToken - warningToken);

}

@Override  
public boolean canPass(Node node, int acquireCount) {  
    return canPass(node, acquireCount, false);  
}

@Override  
public boolean canPass(Node node, int acquireCount, boolean prioritized) {

            long passQps = (long) node.passQps();

            long previousQps = (long) node.previousPassQps();  
    syncToken(previousQps);

                    long restToken = storedTokens.get();  
    if (restToken >= warningToken) {  
        long aboveToken = restToken - warningToken;

                                double warningQps = Math.nextUp(1.0 / (aboveToken * slope + 1.0 / count));  
        if (passQps + acquireCount <= warningQps) {  
            return true;  
        }  
    } else {  
        if (passQps + acquireCount <= count) {  
            return true;  
        }  
    }

    return false;  
}

protected void syncToken(long passQps) {  
    long currentTime = TimeUtil.currentTimeMillis();  
    currentTime = currentTime - currentTime % 1000;  
    long oldLastFillTime = lastFilledTime.get();  
    if (currentTime <= oldLastFillTime) {  
        return;  
    }

    long oldValue = storedTokens.get();  
    long newValue = coolDownTokens(currentTime, passQps);

    if (storedTokens.compareAndSet(oldValue, newValue)) {  
        long currentValue = storedTokens.addAndGet(0 - passQps);  
        if (currentValue < 0) {  
            storedTokens.set(0L);  
        }  
        lastFilledTime.set(currentTime);  
    }

}

private long coolDownTokens(long currentTime, long passQps) {  
    long oldValue = storedTokens.get();  
    long newValue = oldValue;

                    if (oldValue < warningToken) {  
        newValue = (long)(oldValue + (currentTime - lastFilledTime.get()) * count / 1000);  
    } else if (oldValue > warningToken) {  
        if (passQps < (int)count / coldFactor) {  
            newValue = (long)(oldValue + (currentTime - lastFilledTime.get()) * count / 1000);  
        }  
    }  
    return Math.min(newValue, maxToken);  
}

WarmUpController 是用令牌桶算法实现的，还设计了预热机制。