算法详解
计数算法
说明
技术算法，为最简单的限流算法。
核心思想是，每隔一段时间，，为计数器设定最大值，请求一次，计数器数量减一，如果计数器为0，则拒绝请求。
图示流程
适用场景
虽然此算法是大多数人第一个想到，可以限流的算法，但是不推荐使用此算法。
因为，此算法有个致命性的问题，如果1秒允许的访问次数为100，前0.99秒内没有任何请求，在最后0.01秒内，出现了200个请求，则这200个请求，都会获取调用许可，给程序带来一次请求的高峰。
如下图所示：
/**
* 计数法
*/
public class CountLimiter {
/**
* 执行区间(毫秒)
*/
private int secondMill;
/**
* 区间内计数多少次
*/
private int maxCount;
/**
* 当前计数
*/
private int currentCount;
/**
* 上次更新时间（毫秒）
*/
private long lastUpdateTime;
public CountLimiter(int second, int count) {
if (second <= 0 || count <= 0) {
throw new IllegalArgumentException("second and time must by positive");
}
this.secondMill = second * 1000;
this.maxCount = count;
this.currentCount = this.maxCount;
this.lastUpdateTime = System.currentTimeMillis();
}
/**
* 刷新计数器
*/
private void refreshCount() {
long now = System.currentTimeMillis();
if ((now - this.lastUpdateTime) >= secondMill) {
this.currentCount = maxCount;
this.lastUpdateTime = now;
}
}
/**
* 获取授权
* @return
*/
public synchronized boolean tryAcquire() {
// 刷新计数器
this.refreshCount();
if ((this.currentCount - 1) >= 0) {
this.currentCount--;
return true;
} else {
return false;
}
}
public static void main(String[] args) throws InterruptedException {
CountLimiter countLimiter = new CountLimiter(1,2);
for (int i = 0; i < 10; i++) {
System.out.println(LocalDateTime.now() + " " + countLimiter.tryAcquire());
TimeUnit.MILLISECONDS.sleep(200);
}
}
}
漏桶算法
说明
漏桶算法称为leaky bucket，可限制指定时间内的最大流量，如限制60秒内，最多允许100个请求。
其中接受请求的速度是不恒定的（水滴入桶），处理请求的速度是恒定的（水滴出桶）。
算法总体描述如下：
1、有个固定容量的桶B（指定时间区间X，允许的的最大流量B），如60秒内最多允许100个请求，则B为100，X为60。
2、有水滴流进来（有请求进来），桶里的水+1。
3、有水滴流出去（执行请求对应的业务），桶里的水-1（业务方法，真正开始执行=>这是保证漏桶匀速处理业务的根本）水滴流出去的速度是匀速的，流速为B/X,1毫秒100/60次，约1毫秒0.00167次，精度可根据实际情况自己控制）。
4、水桶满了后（60秒内请求达到了100次），水滴无法进入水桶，请求被拒绝。
图示
实际开发中，漏桶的使用方式可参考下图：
需注意，水滴滴落的时候，才开始执行业务代码，而不是水滴进桶的时候，去执行业务代码。
业务代码的执行方式，个人认为有如下两种：
同步执行
1、调用方请求时，如水滴可以放入桶中，调用方所在的线程“阻塞”
2、水滴漏出时，唤醒调用方线程，调用方线程，执行具体业务
异步执行
1、调用方请求时，如水滴可以放入桶中，调用方所在的线程收到响应，方法将异步执行
2、水滴漏出时，水桶代理执行具体业务
网上很多滴桶的实现代码，在水滴进桶的时候，这样会导致业务代码，无法匀速地执行，仍然对被调用的接口有一瞬间流量的冲击（和令牌桶算法的最终实现效果一样）。
适用场景
水桶的进水速度是不可控的，有可能一瞬间有大量的请求进入水桶。处理请求的速度是恒定的（滴水的时候处理请求）。
此算法，主要应用于自己的服务，调用外部接口。以均匀的速度调用外部接口，防止对外部接口的压力过大，而影响外部系统的稳定性。如果影响了别人的系统，力过大，而影响外部系统的稳定性。如果影响了别人的系统，、
代码
本实例代码的实现，在水滴滴下，执行具体业务代码时,采用同步执行的方式。即唤醒调用方的线程，让"调用者"所属的线程去执行具体业务代码，去调用接口。
/**
* 漏桶算法
*/
public class LeakyBucketLimiter {
/**
* 漏桶流出速率(多少纳秒执行一次)
*/
private long outflowRateNanos;
/**
* 漏桶容器
*/
private volatile BlockingQueue<Drip> queue;
/**
* 滴水线程
*/
private Thread outflowThread;
/**
* 水滴
*/
private static class Drip {
/**
* 业务主键
*/
private String busId;
/**
* 水滴对应的调用者线程
*/
private Thread thread;
public Drip(String busId, Thread thread) {
this.thread = thread;
}
public String getBusId() {
return this.busId;
}
public Thread getThread() {
return this.thread;
}
}
/**
* @param second 秒
* @param time 调用次数
*/
public LeakyBucketLimiter(int second, int time) {
if (second <= 0 || time <= 0) {
throw new IllegalArgumentException("second and time must by positive");
}
//多少纳秒执行一次
outflowRateNanos = TimeUnit.SECONDS.toNanos(second) / time;
queue = new LinkedBlockingQueue<>(time);
outflowThread = new Thread(() -> {
while (true) {
Drip drip = null;
try {
// 阻塞，直到从桶里拿到水滴
drip = queue.take();
} catch (Exception e) {
e.printStackTrace();
}
if (drip != null && drip.getThread() != null) {
// 唤醒阻塞的水滴里面的线程
LockSupport.unpark(drip.getThread());
}
// 休息一段时间，开始下一次滴水
LockSupport.parkNanos(this, outflowRateNanos);
}
}, "漏水线程");
outflowThread.start();
}
/**
* 业务请求
*
* @return
*/
public boolean acquire(String busId) {
Thread thread = Thread.currentThread();
Drip drip = new Drip(busId, thread);
//添加失败返回false。
if (this.queue.offer(drip)) {
LockSupport.park();
return true;
} else {
return false;
}
}
public static void main(String[] args) throws InterruptedException {
//1秒限制执行1次
LeakyBucketLimiter leakyBucketLimiter = new LeakyBucketLimiter(5, 2);
for (int i = 0; i < 10; i++) {
new Thread(new Runnable() {
@Override
public void run() {
String busId = "[业务ID：" + LocalDateTime.now().toString() + "]";
if (leakyBucketLimiter.acquire(busId)) {
System.out.println(LocalDateTime.now() + " " + Thread.currentThread().getName() + "：调用外部接口...成功：" + busId);
} else {
System.out.println(LocalDateTime.now() + " " + Thread.currentThread().getName() + "：调用外部接口...失败：" + busId);
}
}
}, "测试线程-" + i).start();
TimeUnit.MILLISECONDS.sleep(500);
}
}
}
令牌桶算法
说明
令牌桶算法，主要是匀速地增加可用令牌，令牌数因为桶的限制有数量上限。
请求拿到令牌，相当于拿到授权，，即可进行相应的业务操作。
图示
适用场景
和漏桶算法比，有可能导致短时间内的请求数上升（因为拿到令牌后，就可以访问接口，有可能一瞬间将所有令牌拿走），但是不会有计数算法那样高的峰值（因为令牌数量是匀速增加的）。
一般自己调用自己的接口，接口会有一定的伸缩性，令牌桶算法，主要用来保护自己的服务器接口。
/**
* 令牌桶限流算法
*/
public class TokenBucketLimiter {
/**
* 桶的大小
*/
private double bucketSize;
/**
* 桶里的令牌数
*/
private double tokenCount;
/**
* 令牌增加速度(每毫秒)
*/
private double tokenAddRateMillSecond;
/**
* 上次更新时间（毫秒）
*/
private long lastUpdateTime;
/**
* @param second 秒
* @param time 调用次数
*/
public TokenBucketLimiter(double second, double time) {
if (second <= 0 || time <= 0) {
throw new IllegalArgumentException("second and time must by positive");
}
// 桶的大小
this.bucketSize = time;
// 桶里的令牌数
this.tokenCount = this.bucketSize;
// 令牌增加速度(每毫秒)
this.tokenAddRateMillSecond = time / second / 1000;
// 上次更新时间（毫秒）
this.lastUpdateTime = System.currentTimeMillis();
}
/**
* 刷新桶内令牌数(令牌数不得超过桶的大小)
* 计算“上次刷新时间”到“当前刷新时间”中间，增加的令牌数
*/
private void refreshTokenCount() {
long now = System.currentTimeMillis();
this.tokenCount = Math.min(this.bucketSize, this.tokenCount + ((now - this.lastUpdateTime) * this.tokenAddRateMillSecond));
this.lastUpdateTime = now;
}
/**
* 尝试拿到权限
*
* @return
*/
public synchronized boolean tryAcquire() {
// 刷新桶内令牌数
this.refreshTokenCount();
if ((this.tokenCount - 1) >= 0) {
// 如果桶中有令牌，令牌数-1
this.tokenCount--;
return true;
} else {
// 桶中已无令牌
return false;
}
}
public static void main(String[] args) throws InterruptedException {
TokenBucketLimiter leakyBucketLimiter = new TokenBucketLimiter(2, 1);
for (int i = 0; i <= 10; i++) {
System.out.println(LocalDateTime.now() + " " + leakyBucketLimiter.tryAcquire());
TimeUnit.SECONDS.sleep(1);
}
}
}
