1、场景
程序中经常需要对接口进行限流，防止访问量太大，导致程序崩溃。
常用的算法有：计数算法、漏桶算法、令牌桶算法，最常用的算法是后面两种。
2、算法详解
2.1 计数算法
2.1.1 说明
技术算法，为最简单的限流算法。
核心思想是，每隔一段时间，为计数器设定最大值，请求一次，计数器数量减一，如果计数器为0，则拒绝请求。
计数器算法图示：
2.1.2 适用场景
虽然此算法是大多数人第一个想到可以限流的算法，但是不推荐使用此算法。
因为，此算法有个致命性的问题，如果1秒允许的访问次数为100，前0.99秒内没有任何请求，在最后0.01秒内，出现了200个请求，则这200个请求，都会获取调用许可，给程序带来一次请求的高峰。
如下图所示：计数器算法缺点
2.1.3 代码
?| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 |
import java.time.localdatetime;
import java.util.concurrent.timeunit;
/**
* 计数器限流器
*/
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;
}
}
}
|
测试方法：
?| 1 2 3 4 5 6 7 8 |
public static void main(string[] args) throws exception {
// 1秒限制执行2次
countlimiter countlimiter = new countlimiter(1, 2);
for (int i = 0; i < 10; i++) {
system.out.println(localdatetime.now() + " " + countlimiter.tryacquire());
timeunit.milliseconds.sleep(200);
}
}
|
执行结果：
?| 1 2 3 4 5 6 7 8 9 10 |
2021-05-31t22:01:08.660 true
2021-05-31t22:01:08.868 true
2021-05-31t22:01:09.074 false
2021-05-31t22:01:09.275 false
2021-05-31t22:01:09.485 false
2021-05-31t22:01:09.698 true
2021-05-31t22:01:09.901 true
2021-05-31t22:01:10.104 false
2021-05-31t22:01:10.316 false
2021-05-31t22:01:10.520 false
|
2.2 漏桶算法
2.2.1 说明
漏桶算法称为leaky bucket，可限制指定时间内的最大流量，如限制60秒内，最多允许100个请求。
其中接受请求的速度是不恒定的（水滴入桶），处理请求的速度是恒定的（水滴出桶）。
算法总体描述如下：
- 有个固定容量的桶b（指定时间区间x，允许的的最大流量b），如60秒内最多允许100个请求，则b为100，x为60。
- 有水滴流进来（有请求进来），桶里的水+1。
-
有水滴流出去（执行请求对应的业务），桶里的水-1
（业务方法，真正开始执行=>这是保证漏桶匀速处理业务的根本），水滴流出去的速度是匀速的，流速为b/x（1毫秒100/60次，约1毫秒0.00167次，精度可根据实际情况自己控制） - 水桶满了后（60秒内请求达到了100次），水滴无法进入水桶，请求被拒绝
2.2.2 漏桶算法图示
实际开发中，漏桶的使用方式可参考下图：
注意：水滴滴落的时候，才开始执行业务代码，而不是水滴进桶的时候，去执行业务代码。
业务代码的执行方式，个人认为有如下两种：
同步执行：
- 调用方请求时，如水滴可以放入桶中，调用方所在的线程“阻塞”
- 水滴漏出时，唤醒调用方线程，调用方线程，执行具体业务
异步执行：
- 调用方请求时，如水滴可以放入桶中，调用方所在的线程收到响应，方法将异步执行
- 水滴漏出时，水桶代理执行具体业务
网上很多滴桶的实现代码，在水滴进桶的时候，就去执行业务代码了。这样会导致业务代码，无法匀速地执行，仍然对被调用的接口有一瞬间流量的冲击（和令牌桶算法的最终实现效果一样）。
2.2.3 适用场景
水桶的进水速度是不可控的，有可能一瞬间有大量的请求进入水桶。处理请求的速度是恒定的（滴水的时候处理请求）。
此算法，主要应用于自己的服务，调用外部接口。以均匀的速度调用外部接口，防止对外部接口的压力过大，而影响外部系统的稳定性。如果影响了别人的系统，接口所在公司会来找你喝茶。
漏桶算法，主要用来保护别人的接口。
2.2.4 代码
本实例代码的实现，在水滴滴下，执行具体业务代码时，采用同步执行的方式。即唤醒调用方的线程，让"调用者"所属的线程去执行具体业务代码，去调用接口。
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 |
import java.net.sockettimeoutexception;
import java.time.localdatetime;
import java.util.queue;
import java.util.uuid;
import java.util.concurrent.arrayblockingqueue;
import java.util.concurrent.blockingqueue;
import java.util.concurrent.linkedblockingqueue;
import java.util.concurrent.timeunit;
import java.util.concurrent.locks.locksupport;
/**
* 漏桶算法
*/
public class leakybucketlimiterutil {
/**
* 漏桶流出速率(多少纳秒执行一次)
*/
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 leakybucketlimiterutil(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);
if (this.queue.offer(drip)) {
locksupport.park();
return true;
} else {
return false;
}
}
}
|
测试代码如下：
?| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 |
public static void main(string[] args) throws exception {
// 1秒限制执行1次
leakybucketlimiterutil leakybucketlimiter = new leakybucketlimiterutil(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);
}
}
|
执行结果如下：
?| 1 2 3 4 5 6 7 8 9 10 |
2021-05-31t20:52:52.297 测试线程-0：调用外部接口...成功：[业务id：2021-05-31t20:52:52.295]
2021-05-31t20:52:53.782 测试线程-3：调用外部接口...失败：[业务id：2021-05-31t20:52:53.782]
2021-05-31t20:52:54.286 测试线程-4：调用外部接口...失败：[业务id：2021-05-31t20:52:54.286]
2021-05-31t20:52:54.799 测试线程-1：调用外部接口...成功：[业务id：2021-05-31t20:52:52.761]
2021-05-31t20:52:55.300 测试线程-6：调用外部接口...失败：[业务id：2021-05-31t20:52:55.300]
2021-05-31t20:52:55.806 测试线程-7：调用外部接口...失败：[业务id：2021-05-31t20:52:55.806]
2021-05-31t20:52:56.307 测试线程-8：调用外部接口...失败：[业务id：2021-05-31t20:52:56.307]
2021-05-31t20:52:56.822 测试线程-9：调用外部接口...失败：[业务id：2021-05-31t20:52:56.822]
2021-05-31t20:52:57.304 测试线程-2：调用外部接口...成功：[业务id：2021-05-31t20:52:53.271]
2021-05-31t20:52:59.817 测试线程-5：调用外部接口...成功：[业务id：2021-05-31t20:52:54.799]
|
2.3 令牌桶算法
2.3.1 说明
令牌桶算法，主要是匀速地增加可用令牌，令牌数因为桶的限制有数量上限。
请求拿到令牌，相当于拿到授权，即可进行相应的业务操作。
2.3.2 令牌桶算法图示
2.3.3 适用场景
和漏桶算法比，有可能导致短时间内的请求数上升（因为拿到令牌后，就可以访问接口，有可能一瞬间将所有令牌拿走），但是不会有计数算法那样高的峰值（因为令牌数量是匀速增加的）。
一般自己调用自己的接口，接口会有一定的伸缩性，令牌桶算法，主要用来保护自己的服务器接口。
2.3.4 代码
代码实现如下：
?| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 |
import java.time.localdatetime;
import java.util.concurrent.timeunit;
/**
* 令牌桶限流算法
*/
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;
}
}
}
|
测试代码：
?| 1 2 3 4 5 6 7 8 |
public static void main(string[] args) throws exception{
// 2秒执行1次
tokenbucketlimiter leakybucketlimiter = new tokenbucketlimiter(2, 1);
for (int i = 0; i < 10; i++) {
system.out.println(localdatetime.now() + " " + leakybucketlimiter.tryacquire());
timeunit.seconds.sleep(1);
}
}
|
执行结果如下：
?| 1 2 3 4 5 6 7 8 9 10 |
2021-05-31t21:38:34.560 true
2021-05-31t21:38:35.582 false
2021-05-31t21:38:36.588 true
2021-05-31t21:38:37.596 false
2021-05-31t21:38:38.608 true
2021-05-31t21:38:39.610 false
2021-05-31t21:38:40.615 true
2021-05-31t21:38:41.627 false
2021-05-31t21:38:42.641 true
2021-05-31t21:38:43.649 false
|
2.3.5 第三方工具类
可以使用guava中的ratelimiter来实现令牌桶的限流功能。
maven依赖如下：
?| 1 2 3 4 5 |
<dependency>
<groupid>com.google.guava</groupid>
<artifactid>guava</artifactid>
<version>30.1.1-jre</version>
</dependency>
|
直接获取令牌（true为获取到令牌，false为获取失败）：
?| 1 2 3 4 5 6 7 |
ratelimiter ratelimiter = ratelimiter.create(2);
boolean acquireresule = ratelimiter.tryacquire();
if (acquireresule) {
system.out.println("获取令牌：成功");
} else {
system.out.println("获取令牌：失败");
}
|
等待尝试获取令牌（阻塞当前线程，直到获取到令牌）：
?| 1 2 3 4 |
ratelimiter ratelimiter = ratelimiter.create(2);
// 阻塞获取令牌
double waitcount = ratelimiter.acquire();
system.out.println("阻塞等待时间：" + waitcount);
|
以上就是java限流算法详细的详细内容，更多关于java限流算法的资料请关注服务器之家其它相关文章！希望大家以后多多支持服务器之家！
原文链接：https://www.jianshu.com/p/d1c456399677
如果您对该产品感兴趣，请填写办理（客服微信：xiaoxiongyidong）
发表评论
◎欢迎参与讨论，请在这里发表您的看法、交流您的观点。