漏桶算法
漏桶算法介绍
漏桶算法，又称leaky bucket。
从图中我们可以看到，整个算法其实十分简单。首先，我们有一个固定容量的桶，有水流进来，也有水流出去。对于流进来的水来说，我们无法预计一共有多少水会流进来，也无法预计水流的速度。但是对于流出去的水来说，这个桶可以固定水流出的速率。而且，当桶满了之后，多余的水将会溢出。
我们将算法中的水换成实际应用中的请求，我们可以看到漏桶算法天生就限制了请求的速度。当使用了漏桶算法，我们可以保证接口会以一个常速速率来处理请求。所以漏桶算法天生不会出现临界问题。
漏桶算法的实现
漏桶算法具体的实现代码如下：
package com.morris.user.demo;
import lombok.extern.slf4j.Slf4j;
import java.util.Random;
import java.util.concurrent.TimeUnit;
/**
* 漏桶算法
*/
@Slf4j
public class LeakyBucketDemo {
public static void main(String[] args) {
LeakyBucket sideWindow = new LeakyBucket(2);
for (int i = 0; i < 30; i++) {
int finalI = i;
new Thread(() -> {
try {
TimeUnit.MILLISECONDS.sleep(new Random().nextInt(30000));
} catch (InterruptedException e) {
e.printStackTrace();
}
log.info(finalI + "--> " + sideWindow.canPass());
}).start();
}
}
public static class LeakyBucket {
// 桶的容量
private static final long CAPACITY = 10;
// 水漏出的速度(每秒系统能处理的请求数)
private final long rate;
// 当前水量(当前累积请求数)
private double water = 0;
// 上一次请求的时间
private long lastRequestTime = System.currentTimeMillis();
/**
* @param maxCount 1秒内允许的最大请求数
*/
public LeakyBucket(int maxCount) {
// 每秒允许的最大请求数
this.rate = maxCount;
}
public synchronized boolean canPass() {
long now = System.currentTimeMillis();
// 去除从lastRequestTime~now时间段应该从桶中漏出的水
this.water = Math.max(0, water - (now - lastRequestTime) * 1.0D / 1000 * this.rate);
this.lastRequestTime = now;
if (this.water + 1 > CAPACITY) {
// 桶满了
return false;
}
this.water++;
return true;
}
}
}
漏桶算法的实现2
漏桶算法的另一种具体实现代码如下：
package com.morris.user.demo;
import lombok.extern.slf4j.Slf4j;
import java.util.Random;
import java.util.concurrent.TimeUnit;
/**
* 漏桶算法
*/
@Slf4j
public class LeakyBucketDemo2 {
public static void main(String[] args) {
LeakyBucket sideWindow = new LeakyBucket(2);
for (int i = 0; i < 30; i++) {
int finalI = i;
new Thread(() -> {
try {
TimeUnit.MILLISECONDS.sleep(new Random().nextInt(3000));
} catch (InterruptedException e) {
e.printStackTrace();
}
log.info(finalI + "--> " + sideWindow.canPass());
}).start();
}
}
public static class LeakyBucket {
// 上一次请求的时间
private long lastRequestTime = System.currentTimeMillis();
// 两次请求之间的最小时间间隔
private final int interval;
public LeakyBucket(int maxCount) {
this.interval = 1000 / maxCount;
}
public synchronized boolean canPass() {
long now = System.currentTimeMillis();
// 两次请求之间的时间间隔
if (now - lastRequestTime >= interval) {
lastRequestTime = now;
return true;
}
return false;
}
}
}
漏桶算法的优缺点
优点：
- 可以平滑限制请求的处理速度，避免瞬间请求过多导致系统崩溃或者雪崩。
- 可以控制请求的处理速度，使得系统可以适应不同的流量需求，避免过载或者过度闲置。
- 可以通过调整桶的大小和漏出速率来满足不同的限流需求，可以灵活地适应不同的场景。
缺点：
- 需要对请求进行缓存，会增加服务器的内存消耗。
- 对于流量波动比较大的场景，需要较为灵活的参数配置才能达到较好的效果。
- 但是面对突发流量的时候，漏桶算法还是循规蹈矩地处理请求，这不是我们想看到的啦。流量变突发时，我们肯定希望系统尽量快点处理请求，提升用户体验嘛。
漏桶算法的缺点主要是由于服务和漏下来的请求之间没有交互，要么服务太忙，要么服务太闲，这就引出了令牌桶算法。
漏桶算法在sentinel中的应用
漏桶算法主要用于流控规则中流控效果为排队等待。
com.alibaba.csp.sentinel.slots.block.flow.controller.RateLimiterController#canPass(com.alibaba.csp.sentinel.node.Node, int, boolean)
public boolean canPass(Node node, int acquireCount, boolean prioritized) {
// Pass when acquire count is less or equal than 0.
if (acquireCount <= 0) {
return true;
}
// Reject when count is less or equal than 0.
// Otherwise,the costTime will be max of long and waitTime will overflow in some cases.
if (count <= 0) {
return false;
}
// 当前时间
long currentTime = TimeUtil.currentTimeMillis();
// 计算两个请求中间允许的时间间隔
// Calculate the interval between every two requests.
long costTime = Math.round(1.0 * (acquireCount) / count * 1000);
// Expected pass time of this request.
// 下一次请求期待的访问时间
long expectedTime = costTime + latestPassedTime.get();
if (expectedTime <= currentTime) {
// 当前时间大于下一次请求期待的访问时间，允许通过
// Contention may exist here, but it's okay.
latestPassedTime.set(currentTime);
return true;
} else {
// Calculate the time to wait.
// 计算当前线程需要等待的时间
long waitTime = costTime + latestPassedTime.get() - TimeUtil.currentTimeMillis();
if (waitTime > maxQueueingTimeMs) {
// 等待时间大于设置的最大的等待超时时间
return false;
} else {
// latestPassedTime+costTime就是下一次请求期待的访问时间
long oldTime = latestPassedTime.addAndGet(costTime);
try {
waitTime = oldTime - TimeUtil.currentTimeMillis();
if (waitTime > maxQueueingTimeMs) {
// 说明latestPassedTime被另一个线程改了
// 把时间改回去，然后请求不允许通过
latestPassedTime.addAndGet(-costTime);
return false;
}
// in race condition waitTime may <= 0
if (waitTime > 0) {
// 线程休眠，排队等待
Thread.sleep(waitTime);
}
return true;
} catch (InterruptedException e) {
}
}
}
return false;
}
