在开发高并发系统时有三把利器用来保护系统：缓存、降级、 限流 , 今天我们就谈谈
限流缓存：缓存的目的是提升系统访问速度和增大系统处理容量 降级：降级是当服务器压力剧增的情况下，根据当前业务情况及流量对一些服务和页面有策略的降级，以此释放服务器资源以保证核心任务的正常运行 限流：限流的目的是通过对并发访问/请求进行限速，或者对一个时间窗口内的请求进行限速来保护系统，一旦达到限制速率则可以拒绝服务、排队或等待、降级等处理
我下面算法的实现基本上都用到了定时器
Timer, 其实关于时间的也可以不用定时器, 可以看看Guava的RateLimiter, 定时器的好处是我不用处理时间逻辑 , 但是需要消耗一个线程去执行逻辑 , 当逻辑算力压力过大会线程处理不过来,效果不好 , 可以使用一下ScheduledThreadPoolExecutor线程池来执行,降低压力同时还大量使用了
队列数据结构 ,是因为生产者消费者模型大多需要队列, 先进先出的特点第一节是环境搭建 , 写出需求 ,和接口要求 , 和测试用例 ,后面四节就是基本算法
1. 环境搭建
我们模拟Filter#doFilter 接口进行测试 , 全部实现 AbstractLimiter#limit方法
Filter 实现
public interface Filter { default public void init() { } public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain); default public void destroy() { } } 复制代码
FilterChain 实现
public interface FilterChain { void doFilter(ServletRequest request, ServletResponse response); } 复制代码
ServletRequest 实现
public class ServletRequest { private String msg; public String getMsg() { return msg; } public void setMsg(String msg) { this.msg = msg; } @Override public String toString() { return "ServletRequest{" + "msg='" + msg + '\'' + '}'; } public ServletRequest(String msg) { this.msg = msg; } } 复制代码
ServletResponse 实现
public class ServletResponse { } 复制代码
AbstractLimiter 实现
public abstract class AbstractLimiter { /** * 最大流量 */ protected final int MAX_FlOW; /** * 构造器 , 输入每秒最大流量 * @param MAX_FlOW 最大流量 */ public AbstractLimiter(int MAX_FlOW) { this.MAX_FlOW = MAX_FlOW; } /** * 具体实现的方法 * @param request 请求 * @param response 响应 * @param chain 执行 */ public abstract void limit(ServletRequest request, ServletResponse response, FilterChain chain); } 复制代码
Demo 测试类
public class Demo { @Test public void test() { // 过滤器 Filter filter = new Filter() { AbstractLimiter limit = null; @Override public void init() { // 入口 ,我们都是每秒限制 100个请求 limit = new LeakyBucketLimiter(100); } @Override public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) { limit.limit(request, response, chain); } }; // 过滤器初始化 filter.init(); // 计时器 long start = System.currentTimeMillis(); // 计数器 AtomicInteger integer = new AtomicInteger(0); ExecutorService pool = Executors.newFixedThreadPool(10); // 模拟4000次请求 IntStream.range(0, 4000).forEach(e -> { try { // 模拟请求延迟 TimeUnit.MILLISECONDS.sleep(1); } catch (InterruptedException e1) { // } // 多线程执行 pool.execute(()->{ filter.doFilter(new ServletRequest("" + e), new ServletResponse(), new FilterChain() { @Override public void doFilter(ServletRequest request, ServletResponse response) { // 回调接口 integer.incrementAndGet(); System.out.println("请求 : "+request.getMsg() + " 通过, 执行线程 "+Thread.currentThread().getName()); } }); }); }); System.out.println("总耗时" + (System.currentTimeMillis() - start)); System.out.println("一共通过 : " + integer.get()); } } 复制代码
2. 计数器算法
计数器算法(Counter)顾明思议就是一个计数器 , 比如我每秒可以通过100个请求 , 我呢每进来一个请求, 我就将计数器+1 , 当计数器到达了100,此时我就不让请求过去 , 但是他存在一个问题 : 比如我第999ms 的时候过来100个请求 , 当刚刚过了1000ms的时候初始化了,但是又来了100个请求 , 此时就会发生实际上在这0.1S的时候处理了200个请求 , 严重超载了 , 此时服务器处理不了而全部都请求超时了....
public class CounterLimiter extends AbstractLimiter { private static final Integer initFlow = 0; private final AtomicInteger flow; public CounterLimiter(int MAX_FlOW) { super(MAX_FlOW); // 初始化计数器 flow = new AtomicInteger(initFlow); new Timer().schedule(new TimerTask() { @Override public void run() { // 每1000ms初始化一次 flow.set(initFlow); } }, 0, 1000); } public void limit(ServletRequest request, ServletResponse response, FilterChain chain) { // 比较是否超载 if (flow.get() < MAX_FlOW) { // 通过 : 计数器+1 flow.incrementAndGet(); chain.doFilter(request, response); } } } 复制代码
3. 滑动窗口算法
滑动窗口算法(Rolling - Window)可以说是计数器算法的一种改进 , 他呢 , 将计算器细分了, 比如我将1S的 1000ms 细分为10个 100ms , 我们就有10个计数器 , 比如上面的问题 , 999ms和1000ms的问题, 由于我们是连续的, 此时1000ms进来的我也算进去了, 此时就不会出现那种情况 ,当我们的颗粒度越高 , 此时所计算的资源会越多,也会越精确 , 其实对比
Hystrix和sentinel都是这种思想, 滑动窗口算法 , 主要是考虑的计算资源少的问题 ,我的算法并不是最优 ,其实不需要使用
ArrayBlockingQueue去维护滑块 , 由于我们是单个线程去执行并不会出现多线程问题, 其实可以使用LinkedList来模拟队列 , 还有其他点也可以看一下
public class RollingWindowFilter extends AbstractLimiter { /** * 我们的滑动窗口对象,包含多个窗口 */ private final Slider slider; /** * 程序中暴露的唯一一个计数器,可以称之为当前窗口 */ private AtomicInteger counter; /** * 计数器初始化大小 */ private static final int INIT_SIZE = 0; /** * 比如窗口分为10块,这个代表先进入9块窗口的计算值 , 为什么要引入是因为不浪费计算资源, 好多都是重复计算 */ private final AtomicInteger preCount; /** * 我们默认队列大小是 20 ,其实颗粒度很高了50ms计算一次, 可以重载构造参数调整 * * @param MAX_FlOW 最大流量 */ public RollingWindowFilter(int MAX_FlOW) { super(MAX_FlOW); // 初始化窗口,感觉改名字叫做Windows比较好 .... slider = new Slider(20); // 初始化对象 preCount = new AtomicInteger(INIT_SIZE); new Timer().schedule(new TimerTask() { @Override public void run() { ArrayBlockingQueue<AtomicInteger> queue = slider.blocks; // 当前窗口大小 int size = queue.size(); /** * 初始化窗口长度 */ if (size < slider.capacity) { try { /** * 计算前面窗口的计数器总和 * * 这里其实由多线程的并发问题 ,其实可以设置一个标识符来表示完成与否 .. 我懒得改了 ,或者你就大量实例化对象,不用我这个单一对象 */ preCount.set(INIT_SIZE); if (size > 0) { queue.forEach(e -> preCount.addAndGet(e.get())); } // 新建一个计数器, 放入对应的滑块 ,其实就是队尾 counter = new AtomicInteger(INIT_SIZE); queue.put(counter); } catch (InterruptedException e) { // } } /** * 当窗口长度初始化完成 */ if (size == slider.capacity) { try { // 出局最先进来的那个 queue.take(); // 计算前面窗口的计数器总和 , 有多线程并发问题 preCount.set(INIT_SIZE); queue.forEach(e -> preCount.addAndGet(e.get())); // 新建一个计数器, 放入对应的滑块 ,其实就是队尾 counter = new AtomicInteger(INIT_SIZE); queue.put(counter); } catch (InterruptedException e) { // } } } }, 0, 1000 / slider.capacity); } public void limit(ServletRequest request, ServletResponse response, FilterChain chain) { int cur = counter.get(); int pre = preCount.get(); int sum = cur + pre; if (sum < MAX_FlOW) { counter.incrementAndGet(); chain.doFilter(request, response); } } /** * 滑块组成 , 一个队列维护一个块 , 其实可以用LinkedList来维护 , 我是懒得改 * <p> * 一般内部类来说看JDK源码你会发现都会用private static修饰 ,因为反射不是静态内部类,无法实例化 , 和构造器不加修饰 */ private static class Slider { // 多少个计数器 private final int capacity; // 放置计数器 private final ArrayBlockingQueue<AtomicInteger> blocks; Slider(int capacity) { this.blocks = new ArrayBlockingQueue<>(capacity); this.capacity = capacity; } } } 复制代码
4. 漏桶算法
其实所谓的
漏桶算法(Leaky Bucket),我们想一下 , 有一个入水口和一个出水口 , 我们这俩口控制权在谁那 ,入水口无非就是大量的请求, 出水口就是我们放过的请求 , 所以他是一个生产者 - 消费者模型, 生产者就是请求 , 消费者就是以一定速度我们消费请求 ,漏桶算法可以使
请求流出的速率是均匀的, 不管你多少请求 , 我流出的速率是均匀的 , 当桶满了就溢出 ,没有满加进来就等着被流出去当你看懂我上面的两段话 , 你就理解了下面的代码 , 我的注释十分清晰
public class LeakyBucketLimiter extends AbstractLimiter { /** * 我们的漏斗 */ private final LeakyBucket leakyBucket; /** * 构造器 , 输入每秒最大流量 * * @param MAX_FlOW 最大流量 */ public LeakyBucketLimiter(int MAX_FlOW) { super(MAX_FlOW); this.leakyBucket = new LeakyBucket(MAX_FlOW); } @Override public void limit(ServletRequest request, ServletResponse response, FilterChain chain) { try { // 1. 获取桶当前水的大小 int size = leakyBucket.bucket.size(); // 2. 比较桶里的水是否满了 if (size < leakyBucket.waterSize) { // 没有满我们就将水放进去,其实这里put也行 , offer也行 , 看需求 leakyBucket.bucket.put(new Water(request, response, chain)); } } catch (InterruptedException e) { // } } static class LeakyBucket { /** * 能放多少水,其实就是队列大小 */ final int waterSize; /** * 我们的放水的桶 */ final ArrayBlockingQueue<Water> bucket; public LeakyBucket(int MAX_FlOW) { this.waterSize = MAX_FlOW; bucket = new ArrayBlockingQueue<>(this.waterSize); /** * 模拟消费 , 1S只能过去100个 ,说明 100ms 可以消耗10个, 看你的颗粒度 */ new Timer().schedule(new TimerTask() { @Override public void run() { // 100ms 流出去10个 for (int i = 0; i < (waterSize / 10); i++) { try { // 流出的水 Water water = bucket.take(); // 执行掉 water.chain.doFilter(water.request, water.response); } catch (InterruptedException e) { // } } } }, 0, 100); } } /** * 我们的节点对象, 其实可以称之为 成功注入的水 , 等着被漏桶流出去 */ static class Water { private ServletRequest request; private ServletResponse response; private FilterChain chain; public Water(ServletRequest request, ServletResponse response, FilterChain chain) { this.request = request; this.response = response; this.chain = chain; } } } 复制代码
5. 令牌桶算法
令牌桶算法(Token Bucket)是与漏桶算法相反的思想, 他也是生产者消费者模型,只是角色的互换, 他呢是我们去控制生成 , 请求去执行消费 , 举个栗子 : 比如我们限流100 , 此时我们就每100ms生成10个令牌 , 当令牌数达到100 我们就不生产 了, 当一个请求过来 , 就会去拿掉一个令牌 , 如果拿到了就通过了, 拿不到就拒绝根据这个我们可以和漏桶算法做比较 ,假设都是刚刚开始 , 此时都是100个请求过来 , 令牌桶可能会拒绝掉90个,因为我只生产了10个令牌 ,但是漏桶呢他不会, 他会将100个请求全部放进去慢慢消费 , 是因为我的桶容量是100,可以放进去这么多请求 , 这就是这俩的区别 .... 其实稳定了几乎么区别
生产者消费者模型的思想转换可以更加理清思路 , 模型的选择有时候是解决问题的一个合适的方式令牌桶算法 网上大多都是采用的
Guava的RateLimiter实现的 , 这里我就实现两种 一种是自己实现, 一种是使用RateLimiter,
1. 自己实现的令牌桶
public class TokenBucketLimiter extends AbstractLimiter { /** * 令牌桶 */ private final TokenBucket tokenBucket; /** * 构造器 , 输入每秒最大流量 * * @param MAX_FlOW 最大流量 */ public TokenBucketLimiter(int MAX_FlOW) { super(MAX_FlOW); this.tokenBucket = new TokenBucket(MAX_FlOW); } @Override public void limit(ServletRequest request, ServletResponse response, FilterChain chain) { /** * 这里我们就不使用 take的阻塞思想了 ,直接poll去拉去 ,然后等待5mS , 如果拉去不到直接返回失败 , 其实等待的长了点 */ try { // 尝试去获取一个令牌 Token token = tokenBucket.bucket.poll(5, TimeUnit.MILLISECONDS); // 拿到通过 if (null != token) { chain.doFilter(request, response); } } catch (InterruptedException e) { // } } /** * 令牌桶 */ private static class TokenBucket { /** * 令牌存放的位置 , 用一个队列维护 */ private final ArrayBlockingQueue<Token> bucket; /** * 桶最多存放多少个令牌 */ private final int tokenSize; public TokenBucket(int MAX_FlOW) { this.tokenSize = MAX_FlOW; this.bucket = new ArrayBlockingQueue<>(this.tokenSize); new Timer().schedule(new TimerTask() { @Override public void run() { for (int x = 0; x < (tokenSize / 10); x++) { try { if (bucket.size() < tokenSize) { // 定时放入令牌 bucket.put(new Token()); } } catch (InterruptedException e) { // } } } }, 0, 100); } } /** * 令牌 */ private static class Token { } } 复制代码
2. 基于Guava 的 RateLimiter实现令牌桶
public class GuavaRateLimiter extends AbstractLimiter { /** * 令牌桶 */ private final RateLimiter limiter; /** * 每次需要的令牌个数 */ private static final int ACQUIRE_NUM = 1; /** * 最长等待时间 */ private static final int WAIT_TIME_PER_MILLISECONDS = 5; /** * 构造器 , 输入每秒最大流量 * * @param MAX_FlOW 最大流量 */ public GuavaRateLimiter(final int MAX_FlOW) { super(MAX_FlOW); limiter = RateLimiter.create(MAX_FlOW); } @Override public void limit(ServletRequest request, ServletResponse response, FilterChain chain) { /** * 意思就是 我尝试去获取1个令牌 ,最大等待时间是 5 ms , 其实太长了, 真是开发也就1ms不到 */ boolean flag = limiter.tryAcquire(ACQUIRE_NUM, WAIT_TIME_PER_MILLISECONDS, TimeUnit.MILLISECONDS); if (flag) { chain.doFilter(request, response); } } }
