Java 常见的限流算法详细分析并实现
作者：田埂、 发布时间：2022-01-05 22:11:09
标签：Java,限流算法
为什么要限流
在保证可用的情况下尽可能多增加进入的人数,其余的人在排队等待,或者返回友好提示,保证里面的进行系统的用户可以正常使用，防止系统雪崩。
限流算法
限流算法很多,常见的有三类,分别是 计数器算法 、漏桶算法、令牌桶算法 。
（1）计数器：
在一段时间间隔内，处理请求的最大数量固定，超过部分不做处理。
（2）漏桶：
漏桶大小固定，处理速度固定，但请求进入速度不固定（在突 * 况请求过多时，会丢弃过多的请求）。
（3）令牌桶：
令牌桶的大小固定，令牌的产生速度固定，但是消耗令牌（即请求）速度不固定（可以应对一些某些时间请求过多的情况）；每个请求都会从令牌桶中取出令牌，如果没有令牌则丢弃该次请求。
计数器限流
在一段时间间隔内，处理请求的最大数量固定，超过部分不做处理。
举个例子,比如我们规定对于A接口，我们1分钟的访问次数不能超过100次。
那么我们可以这么做：
在一开 始的时候，我们可以设置一个计数器counter，每当一个请求过来的时候，counter就加1，如果counter的值大于100并且该请求与第一个请求的间隔时间还在1分钟之内，那么说明请求数过多,拒绝访问；
如果该请求与第一个请求的间隔时间大于1分钟，且counter的值还在限流范围内，那么就重置 counter,就是这么简单粗暴。
代码实现：
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
//计数器 限流
public class CounterLimiter {
//起始时间
private static long startTime = System.currentTimeMillis();
//时间间隔1000ms
private static long interval = 1000;
//每个时间间隔内，限制数量
private static long limit = 3;
//累加器
private static AtomicLong accumulator = new AtomicLong();
/**
* true 代表放行，请求可已通过
* false 代表限制，不让请求通过
*/
public static boolean tryAcquire() {
long nowTime = System.currentTimeMillis();
//判断是否在上一个时间间隔内
if (nowTime < startTime + interval) {
//如果还在上个时间间隔内
long count = accumulator.incrementAndGet();
if (count <= limit) {
return true;
} else {
return false;
}
} else {
//如果不在上一个时间间隔内
synchronized (CounterLimiter.class) {
//防止重复初始化
if (nowTime > startTime + interval) {
startTime = nowTime;
accumulator.set(0);
}
}
//再次进行判断
long count = accumulator.incrementAndGet();
if (count <= limit) {
return true;
} else {
return false;
}
}
}
// 测试
public static void main(String[] args) {
//线程池，用于多线程模拟测试
ExecutorService pool = Executors.newFixedThreadPool(10);
// 被限制的次数
AtomicInteger limited = new AtomicInteger(0);
// 线程数
final int threads = 2;
// 每条线程的执行轮数
final int turns = 20;
// 同步器
CountDownLatch countDownLatch = new CountDownLatch(threads);
long start = System.currentTimeMillis();
for (int i = 0; i < threads; i++) {
pool.submit(() ->
{
try {
for (int j = 0; j < turns; j++) {
boolean flag = tryAcquire();
if (!flag) {
// 被限制的次数累积
limited.getAndIncrement();
}
Thread.sleep(200);
}
} catch (Exception e) {
e.printStackTrace();
}
//等待所有线程结束
countDownLatch.countDown();
});
}
try {
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
float time = (System.currentTimeMillis() - start) / 1000F;
//输出统计结果
System.out.println("限制的次数为：" + limited.get() +
",通过的次数为：" + (threads * turns - limited.get()));
System.out.println("限制的比例为：" + (float) limited.get() / (float) (threads * turns));
System.out.println("运行的时长为：" + time + "s");
}
}
计数器限流的不足：
这个算法虽然简单，但是存在临界问题，我们看下图：
从上图中我们可以看到，假设有一个恶意用户，他在0:59时，瞬间发送了100个请求，并且1:00又瞬间发送了100个请求，那么其实这个用户在 1秒里面，瞬间发送了200个请求。
我们刚才规定的是1分钟最多100个请求（规划的吞吐量），也就是每秒钟最多1.7个请求，用户通过在时间窗口的重置节点处突发请求， 可以瞬间超过我们的速率限制。
用户有可能通过算法的这个漏洞，瞬间压垮我们的应用。
漏桶限流
漏桶算法限流的基本原理为：水（对应请求）从进水口进入到漏桶里，漏桶以一定的速度出水（请求放行），当水流入速度过大，桶内的总水量大于桶容量会直接溢出，请求被拒绝。
大致的漏桶限流规则如下：
（1）进水口（对应客户端请求）以任意速率流入进入漏桶。
（2）漏桶的容量是固定的，出水（放行）速率也是固定的。
（3）漏桶容量是不变的，如果处理速度太慢，桶内水量会超出了桶的容量，则后面流入的水滴会溢出，表示请求拒绝。
⭐漏桶算法其实很简单，可以粗略的认为就是注水漏水过程，往桶中以任意速率流入水，以一定速率流出水，当水超过桶容量（capacity）则丢弃，因为桶容量是不变的，保证了整体的速率。
以一定速率流出水，
削峰： 有大量流量进入时,会发生溢出,从而限流保护服务可用
缓冲： 不至于直接请求到服务器, 缓冲压力
代码实现：
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
//漏斗限流
public class LeakBucketLimiter {
//桶的大小
private static long capacity = 10;
//流出速率,每秒两个
private static long rate = 2;
//开始时间
private static long startTime = System.currentTimeMillis();
//桶中剩余的水
private static AtomicLong water = new AtomicLong();
/**
* true 代表放行，请求可已通过
* false 代表限制，不让请求通过
*/
public synchronized static boolean tryAcquire() {
//如果桶的余量问0，直接放行
if (water.get() == 0) {
startTime = System.currentTimeMillis();
water.set(1);
return true;
}
//计算从当前时间到开始时间流出的水，和现在桶中剩余的水
//桶中剩余的水
water.set(water.get() - (System.currentTimeMillis() - startTime) / 1000 * rate);
//防止出现<0的情况
water.set(Math.max(0, water.get()));
//设置新的开始时间
startTime += (System.currentTimeMillis() - startTime) / 1000 * 1000;
//如果当前水小于容量，表示可以放行
if (water.get() < capacity) {
water.incrementAndGet();
return true;
} else {
return false;
}
}
// 测试
public static void main(String[] args) {
//线程池，用于多线程模拟测试
ExecutorService pool = Executors.newFixedThreadPool(10);
// 被限制的次数
AtomicInteger limited = new AtomicInteger(0);
// 线程数
final int threads = 2;
// 每条线程的执行轮数
final int turns = 20;
// 同步器
CountDownLatch countDownLatch = new CountDownLatch(threads);
long start = System.currentTimeMillis();
for (int i = 0; i < threads; i++) {
pool.submit(() ->
{
try {
for (int j = 0; j < turns; j++) {
boolean flag = tryAcquire();
if (!flag) {
// 被限制的次数累积
limited.getAndIncrement();
}
Thread.sleep(200);
}
} catch (Exception e) {
e.printStackTrace();
}
//等待所有线程结束
countDownLatch.countDown();
});
}
try {
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
float time = (System.currentTimeMillis() - start) / 1000F;
//输出统计结果
System.out.println("限制的次数为：" + limited.get() +
",通过的次数为：" + (threads * turns - limited.get()));
System.out.println("限制的比例为：" + (float) limited.get() / (float) (threads * turns));
System.out.println("运行的时长为：" + time + "s");
}
}
漏桶的不足：
漏桶的出水速度固定，也就是请求放行速度是固定的。
漏桶出口的速度固定，不能灵活的应对后端能力提升。比如，通过动态扩容，后端流量从1000QPS提升到1WQPS，漏桶没有办法。
令牌桶限流
令牌桶算法中新请求到来时会从桶里拿走一个令牌，如果桶内没有令牌可拿，就拒绝服务。 当然，令牌的数量也是有上限的。令牌的数量与时间和发放速率强相关，时间流逝的时间越长，会不断往桶里加入越多的令牌，如果令牌发放的速度比申请速度快，令牌桶会放满令牌，直到令牌占满整个令牌桶。
令牌桶限流大致的规则如下：
（1）进水口按照某个速度，向桶中放入令牌。
（2）令牌的容量是固定的，但是放行的速度不是固定的，只要桶中还有剩余令牌，一旦请求过来就能申请成功，然后放行。
（3）如果令牌的发放速度，慢于请求到来速度，桶内就无牌可领，请求就会被拒绝。
总之，令牌的发送速率可以设置，从而可以对突发的出口流量进行有效的应对。
代码实现：
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
//令牌桶
public class TokenBucketLimiter {
//桶的容量
private static long capacity = 10;
//放入令牌的速率,每秒2个
private static long rate = 2;
//上次放置令牌的时间
private static long lastTime = System.currentTimeMillis();
//桶中令牌的余量
private static AtomicLong tokenNum = new AtomicLong();
/**
* true 代表放行，请求可已通过
* false 代表限制，不让请求通过
*/
public synchronized static boolean tryAcquire() {
//更新桶中剩余令牌的数量
long now = System.currentTimeMillis();
tokenNum.addAndGet((now - lastTime) / 1000 * rate);
tokenNum.set(Math.min(capacity, tokenNum.get()));
//更新时间
lastTime += (now - lastTime) / 1000 * 1000;
//桶中还有令牌就放行
if (tokenNum.get() > 0) {
tokenNum.decrementAndGet();
return true;
} else {
return false;
}
}
//测试
public static void main(String[] args) {
//线程池，用于多线程模拟测试
ExecutorService pool = Executors.newFixedThreadPool(10);
// 被限制的次数
AtomicInteger limited = new AtomicInteger(0);
// 线程数
final int threads = 2;
// 每条线程的执行轮数
final int turns = 20;
// 同步器
CountDownLatch countDownLatch = new CountDownLatch(threads);
long start = System.currentTimeMillis();
for (int i = 0; i < threads; i++) {
pool.submit(() ->
{
try {
for (int j = 0; j < turns; j++) {
boolean flag = tryAcquire();
if (!flag) {
// 被限制的次数累积
limited.getAndIncrement();
}
Thread.sleep(200);
}
} catch (Exception e) {
e.printStackTrace();
}
//等待所有线程结束
countDownLatch.countDown();
});
}
try {
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
float time = (System.currentTimeMillis() - start) / 1000F;
//输出统计结果
System.out.println("限制的次数为：" + limited.get() +
",通过的次数为：" + (threads * turns - limited.get()));
System.out.println("限制的比例为：" + (float) limited.get() / (float) (threads * turns));
System.out.println("运行的时长为：" + time + "s");
}
}
令牌桶的好处：
令牌桶的好处之一就是可以方便地应对 突发出口流量（后端能力的提升）。
比如，可以改变令牌的发放速度，算法能按照新的发送速率调大令牌的发放数量，使得出口突发流量能被处理。
来源：https://blog.csdn.net/weixin_51626435/article/details/123561138
0
投稿猜你喜欢
- 双向信号和竞赛(Two-Way Signaling and Races) Monitor.Pulse方法的一个重要特性是它是异步执
- 线程概念进程：启动一个应用程序就叫一个进程。 接着又启动一个应用程序，这叫两个进程。每个进程都有一个独立的内存空间；进程也是程序的一次执行过
一、同步容器 1、Vector——>ArrayList vector 是线程（Thread）同步（Synchron- android电话管理器(TelephonyManger)实例:TelephonyManger是管理电话状态、网络信息的服务类。添加权限：&
- 一、静态代理模式1.1、 代理模式的定义：由于某些原因需要给某对象提供一个代理以控制对该对象的访问。这时，访问对象不适合或者不能直接引用目标
- 两个接口都是继承自Collection.List (inteface) 次序是List 的最重要特点,它确保维护元素特定的顺序. －－Arr
- 哈希 Hash 算法介绍哈希算法也叫散列算法, 不过英文单词都是 Hash, 简单一句话概括, 就是可以把任意长度的输入信息通过算法变换成固
- Android的主线程中执行长时间操作，导致界面无响应，会引起ANR。如果需要执行较长时间的操作，一般会在另一个线程处理，然后将数据转交给主
- 一般要做正圆形图片，只能是正方形的基础上才能实现，否则就变成椭圆了，下面说说如何使长方形的图片生成正圆形图片废话不多说，没图没真相，先上图吧
- 效果自定义密码输入框，项目的一个界面需求，我把这个自定义的输入框提取出来作为这次内容的题目。输入前: 输入后： 输入1个
- 基本环境语言：Java 8 数据库：Oracle ORM 框架：MyBatis 3.4.5需求批量插入数据，数据需要有自增 id。每次插入有
- 混合事务在ORM框架的事务管理器的事务内，使用JdbcTemplate执行SQL是不会纳入事务管理的。下面进行源码分析，看为什么必须要在Da
- 本文实例讲述了Android实现在屏幕上移动图片的方法。分享给大家供大家参考。具体实现方法如下：1. Speed.java文件：packag
- 好了下面进入正题，我们先看一下实现效果吧：下面来介绍一下代码： 本思路就是：1.先到手机中扫描jpeg和png的图片2.获取导图片
- Mybatis 复杂对象resultMap数据对象含有普通属性，一对一对象，一对多对象（2种情况：单一主键和复合主键）下面是resultMa
- 本文实例为大家分享了无限级联下拉列表框的的实现方法，具体内容如下可能有一个树型结构的表，它可能有ID,Name,ParentID,Level
- 引言相信大家在生活中，特别是最近的双十一活动期间，会收到很多短信，而那些短信都有两个特征，第一个是几乎都是垃圾短信，这个特点此处可以忽略不计
- MyBatis多数据源配置(读写分离)首先说明，本文的配置使用的最直接的方式，实际用起来可能会很麻烦。实际应用中可能存在多种结合的情况，你可
- 写在前面在平时的开发之中，我们需要对于数据加载的情况进行展示：空数据网络异常加载中等等情况现在设置页面状态的方式有多种，由于笔者近期一直在使
- 先附spring各版本jar包下载链接http://repo.spring.io/release/org/springframework/s
