服务器之家:专注于服务器技术及软件下载分享
分类导航

PHP教程|ASP.NET教程|Java教程|ASP教程|编程技术|正则表达式|C/C++|IOS|C#|Swift|Android|VB|R语言|JavaScript|易语言|vb.net|

服务器之家 - 编程语言 - Java教程 - java限流算法详细

java限流算法详细

2021-12-14 13:08水煮鱼又失败了 Java教程

这篇文章详细的介绍了java限流算法常用到的算法计数算法、漏桶算法、令牌桶等算法的相关资料,需要的朋友可以参考下文,希望本篇文章能帮助到您

 1、场景

程序中经常需要对接口进行限流,防止访问量太大,导致程序崩溃。

常用的算法有:计数算法、漏桶算法令牌桶算法,最常用的算法是后面两种。

2、算法详解

2.1 计数算法

2.1.1 说明

技术算法,为最简单的限流算法。

核心思想是,每隔一段时间,为计数器设定最大值,请求一次,计数器数量减一,如果计数器为0则拒绝请求

计数器算法图示:

java限流算法详细

2.1.2 适用场景

虽然此算法是大多数人第一个想到可以限流的算法,但是不推荐使用此算法

因为,此算法有个致命性的问题,如果1秒允许的访问次数为100,前0.99秒内没有任何请求,在最后0.01秒内,出现了200个请求,则这200个请求,都会获取调用许可,给程序带来一次请求的高峰。

如下图所示:计数器算法缺点

java限流算法详细

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 漏桶算法图示

实际开发中,漏桶的使用方式可参考下图:

java限流算法详细

注意:水滴滴落的时候,才开始执行业务代码,而不是水滴进桶的时候,去执行业务代码。

业务代码的执行方式,个人认为有如下两种:

同步执行:

  • 调用方请求时,如水滴可以放入桶中,调用方所在的线程“阻塞”
  • 水滴漏出时,唤醒调用方线程,调用方线程,执行具体业务

异步执行:

  • 调用方请求时,如水滴可以放入桶中,调用方所在的线程收到响应,方法将异步执行
  • 水滴漏出时,水桶代理执行具体业务

网上很多滴桶的实现代码,在水滴进桶的时候,就去执行业务代码了。这样会导致业务代码,无法匀速地执行,仍然对被调用的接口有一瞬间流量的冲击(和令牌桶算法的最终实现效果一样)。

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 令牌桶算法图示

java限流算法详细

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

延伸 · 阅读

精彩推荐