限流是对系统的的一种保护措施.即限制流量请求的频率(每秒处理多少个请求).一般来说,当请求流量超过系统的瓶颈时,就丢弃多余的请求流量,保证系统的可用性.要么不放进来,放进来的就保证提供服务
3.1 计数器
概述
计数器采用简单的计数操作,到一段时间节点后自动清零
实现
package com.zhb.algorithm.limit;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
/**
* @author zhuohb
* @date 2021/9/11 3:01 下午
* 限流算法-计数器
*/
public class Counter {
public static void main(String[] args) {
//计数器 这里用信号量实现
final Semaphore semaphore = new Semaphore(3);
//定时器,到点清零
ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
service.scheduleAtFixedRate(() -> semaphore.release(3), 3000, 3000, TimeUnit.MILLISECONDS);
//模拟无数个请求从天而降
while (true) {
try {
semaphore.acquire();
} catch (InterruptedException e) {
e.printStackTrace();
}
//如果准许相应,打印
System.out.println("ok");
}
}
}
结果分析
3个ok为一组呈现,到下一个计数周期之前被阻断
优缺点
- 实现非常简单
- 控制力度过于简略,加入1秒内限制3次,那么3次在前面100毫秒用完,后面的900毫秒将只能处于阻塞状态,白白浪费资源
应用
使用计数器限流的场景较少,因为他的处理逻辑不够灵活.最常见的可能在web的登录密码验证,输入错误次数过多就冻结一段时间的场景.如果网站请求使用计数器,那么恶意攻击者前100毫秒吃掉流量计数,使得后续正常的请求全部阻断
3.2 漏桶算法
概述
漏桶算法将请求缓存在桶中,服务流程匀速处理.超出桶容量的部分丢弃.漏桶算法只要用于保护内部的处理业务,保障其稳定有节奏的处理请求,但是无法根据流量的波动弹性调整响应能力.现实中,类似容纳人数有限的服务大厅开启了固定的服务窗口
补图
实现
package com.zhb.algorithm.limit;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
/**
* @author zhuohb
* @date 2021/9/11 4:11 下午
* 限流算法-漏桶算法
*/
public class Barrel {
public static void main(String[] args) {
//桶,用阻塞队列实现,容量是3
final LinkedBlockingQueue<Integer> queue = new LinkedBlockingQueue<>(3);
//定时器,相当于服务的窗口 2s处理一个
ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
service.scheduleAtFixedRate(() -> System.out.println("处理:" + queue.poll()), 2000, 2000, TimeUnit.MILLISECONDS);
//无数个请求,i理解为请求的编号
int i = 0;
while (true) {
i++;
System.out.println("put:" + i);
try {
//如果是put,会一直等待桶中有空闲位置,不会丢弃
// queue.put(i);
//等待1秒如果进不来桶,就溢出丢弃
queue.offer(i, 1000, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
优缺点
- 有效的挡住了外部的请求,保护了内部的服务不会过载
- 内部服务匀速执行,无法应对流量洪峰,无法做到弹性处理突发任务
- 任务超时溢出时被丢弃,现实中可能需要缓存队列辅助保持一段时间
应用
nginx中的限流是漏桶算法的典型应用,内容有空再补
3.3 令牌桶
概述
令牌桶算法可以认为是漏桶算法的一种升级,它不但可以将流量做一步限制,还可以解决漏桶中无法弹性伸缩处理请求的问题,体现在现实中,类似服务大厅的门口设置门禁卡发放.发送是匀速的,请求较少时候,令牌可以缓存起来,供流量爆发时一次性批量获取使用,而内部服务窗口不设限
补图
实现
package com.zhb.algorithm.limit;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
/**
* @author zhuohb
* @date 2021/9/11 6:43 下午
* 限流算法-令牌桶
*/
public class Token {
public static void main(String[] args) throws Exception {
//令牌桶,信号量实现,容量为3
Semaphore semaphore = new Semaphore(3);
//定时器,1秒一个,匀速颁发令牌
ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
service.scheduleAtFixedRate(() -> {
if (semaphore.availablePermits() < 3) {
semaphore.release();
}
}, 1000, 1000, TimeUnit.MILLISECONDS);
//等待,等候令牌桶存储
Thread.sleep(5);
//模拟洪峰5个请求,前三个迅速响应,后两个排队
for (int i = 0; i < 5; i++) {
semaphore.acquire();
System.out.println("洪峰:" + i);
}
//模拟日常请求,2秒一个
for (int i = 0; i < 3; i++) {
Thread.sleep(1000);
semaphore.acquire();
System.out.println("日常:" + i);
Thread.sleep(1000);
}
//再次洪峰
for (int i = 0; i < 5; i++) {
semaphore.acquire();
System.out.println("洪峰:" + i);
}
//检查令牌桶的数量
for (int i = 0; i < 5; i++) {
Thread.sleep(2000);
System.out.println("剩余令牌:" + semaphore.availablePermits());
}
}
}
应用
springcloud中geteway可以配置令牌桶实现限流控制....有空再补
3.4 滑动窗口
概述
滑动窗口可以理解为细分之后的计数器,计数器粗暴的限定1分钟内的访问次数,而滑动窗口限流将1分钟拆分为多个段,不但要求整个1分钟内请求数小于上线,而且要求每个片段请求数也要小于上线.相当于将原来的计数周期拆分成多个片段,更为精细
补图
实现
package com.zhb.algorithm.limit;
import java.util.LinkedList;
import java.util.Map;
import java.util.TreeMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
/**
* @author zhuohb
* @date 2021/9/11 6:59 下午
* 限流算法-滑动窗口
*/
public class Window {
/**
* 整个窗口的流量上限，超出会被限流
*/
final int totalMax = 5;
/**
* 每片的流量上限，超出同样会被拒绝，可以设置不同的值
*/
final int sliceMax = 5;
/**
* 分多少片
*/
final int slice = 3;
/**
* 窗口，分3段，每段1s，也就是总长度3s
*/
final LinkedList<Long> linkedList = new LinkedList<>();
/**
* 计数器，每片一个key，可以使用HashMap，这里为了控制台保持有序性和可读性，采用TreeMap
*/
Map<Long, AtomicInteger> map = new TreeMap<>();
/**
* 心跳，每1s跳动1次，滑动窗口向前滑动一步，实际业务中可能需要手动控制滑动窗口的时机。
*/
ScheduledExecutorService service = Executors.newScheduledThreadPool(1);
/**
* 获取key值
*
* @return
*/
private Long getKey() {
return System.currentTimeMillis() / 1000;
}
public Window() {
//初始化窗口，当前时间指向的是最末端，前两片其实是过去的2s
long key = getKey();
for (int i = 0; i < slice; i++) {
linkedList.addFirst(key - i);
map.put(key - i, new AtomicInteger(0));
}
//启动心跳任务，窗口根据时间，自动向前滑动，每秒1步
service.scheduleAtFixedRate(() -> {
Long key1 = getKey();
//队尾添加最新的片
linkedList.addLast(key1);
map.put(key1, new AtomicInteger());
//将最老的片移除
map.remove(linkedList.getFirst());
linkedList.removeFirst();
System.out.println("step:" + key1 + ":" + map);
}, 1000, 1000, TimeUnit.MILLISECONDS);
}
/**
* 检查当前时间所在的片是否达到上限
* @return
*/
public boolean checkCurrentSlice() {
long key = getKey();
AtomicInteger integer = map.get(key);
if (integer != null) {
return integer.get() < totalMax;
}
//默认允许访问
return true;
}
/**
* 检查整个窗口所有片的计数之和是否达到上限
*/
public boolean checkAllCount() {
return map.values().stream().mapToInt(AtomicInteger::get).sum() < sliceMax;
}
/**
* 请求来临....
*/
public void req() {
Long key = getKey();
//如果时间窗口未到达当前时间片，稍微等待一下
//其实是一个保护措施，防止心跳对滑动窗口的推动滞后于当前请求
while (linkedList.getLast() < key) {
try {
Thread.sleep(200);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//开始检查，如果未达到上限，返回ok，计数器增加1
//如果任意一项达到上限，拒绝请求，达到限流的目的
//这里是直接拒绝。现实中可能会设置缓冲池，将请求放入缓冲队列暂存
if (checkCurrentSlice() && checkAllCount()) {
//这里的写法是有问题的,200毫秒等待后可能出现跨秒,导致空指针
map.get(key).incrementAndGet();
System.out.println(key + "=ok:" + map);
} else {
System.out.println(key + "=reject:" + map);
}
}
public static void main(String[] args) throws InterruptedException {
Window window = new Window();
//模拟10个离散的请求，相对之间有200ms间隔。会造成总数达到上限而被限流
for (int i = 0; i < 10; i++) {
Thread.sleep(200);
window.req();
}
//等待一下窗口滑动，让各个片的计数器都置零
Thread.sleep(3000);
//模拟突发请求，单个片的计数器达到上限而被限流
System.out.println("---------------------------");
for (int i = 0; i < 10; i++) {
window.req();
}
}
}
