引贴:
在开发高并发系统时有三把利器用来保护系统:缓存、降级和限流
缓存
缓存的目的是提升系统访问速度和增大系统处理容量降级
降级是当服务出现问题或者影响到核心流程时,需要暂时屏蔽掉,待高峰或者问题解决后再打开限流
限流的目的是通过对并发访问/请求进行限速,或者对一个时间窗口内的请求进行限速来保护系统,一旦达到限制速率则可以拒绝服务、排队或等待、降级等处理
RateLimiter
由 guava 提供,常用在削峰控流的场景中。
Java 并发库 的Semaphore信号量控制也可以做到一定的控制: 通过 acquire() 获取一个许可,如果没有就等待,而 release() 释放一个许可 。当然在具体使用的业务场景中,要考虑既然都要限流了,是否线程池的配置合理!RateLimiter的两种模式: SmoothBursty(稳定模式) & SmoothWarmingUp(渐进模式) 。
- 线程安全的,在核心方法中使用synchronized 加锁来控制并发。
- 构造参数里有一个SleepingStopwatch,实际上是TimeUnit.sleep来控制当前请求的阻塞时长。
- 并发请求时,当令牌第一次不够时确定下一次的阻塞时间,下一次才会真正阻塞!!
成员变量
- storedPermits: 当前存储令牌数
- maxPermits: 最大存储令牌数, 应对突然的高并发
- stableIntervalMicros:生成单个令牌需要时间。 eg. 比如说是每秒5个,那间隔时间200ms
- nextFreeTicketMicros :下一次请求可以获取令牌的起始时间 。 由于RateLimiter允许预消费,上次请求预消费令牌后, 下次请求需要等待相应的时间到nextFreeTicketMicros时刻才可以获取令牌
实现原理
以SmoothBursty 为例:
- 在初始化(create)时:
- 初始化SleepingStopwatch,
- 确定好每个令牌在每秒生成的间隔时间stableIntervalMicros,
- 确定好令牌最大数量(maxPermits)(SmoothBursty 模式下默认是1s的量),
- 确定好下一个令牌可供给时间(nextFreeTicketMicros)<初始化值等于构建raterLimiter时的TimeMicros>。
- 每次acquire时:
- 比较请求时间比此时的nextFreeTicketMicros大, 计算这段时间间隔内会生成的令牌,(不超过最大值),确定此时的令牌总量storedPermits; 并更新nextFreeTicketMicros 为当前请求时间
- 计算所需要的令牌是否充足
- 注意: 这里返回的returnValue是此时的nextFreeTicketMicros,(如果当前请求时间 小于 此时的nextFreeTicketMicros,那这个nextFreeTicketMicros是上一个请求处理得到的值)!!
- 如果充足,则直接处理令牌总量 storedPermits = 原storedPermits - 需求量requiredPermits
- 如果不充足,
- 更新nextFreeTicketMicros 的值为 原 nextFreeTicketMicros + 生成差额令牌需要的时间;
- 更新令牌总量 storedPermits = 0;
- 依靠SleepingStopwatch来休眠(实际上是TimeUnit.sleep)指定时长。
SmoothWarmingUp
- 冷启动时会以一个比较大的速率慢慢到平均速率,然后趋于平均速率(梯形下降到平均速率)。
- warmupPeriod * TimeUnit 越大,上升速率越平滑。
最大数则是由初始化时预热时长warmupPeriod 除以间隔时间stableIntervalMicros的值,初始化时预热令牌!
- doSetRate: 在设置初始值时不同。
- 处理等待时间的计算规则上不同:
- 稳定模式SmoothBursty : waitMicros = 差额 * 间隔时长 。调用时将允许一定时间的突发,取决于: 上一次访问距当前访问时间内生成的不超过最大容量的令牌数;
- 渐进模式SmoothWarmingUp: waitMicros = 差额 * 间隔时长 + 对slope的处理。即使差额为0,也需要与slope进行计算,会产生一个浮动值。冷启动时会以一个比较大的速率慢慢到平均速率,然后趋于平均速率。
模式验证
SmoothBursty允许一定程度的突发
package com.noob;import java.lang.reflect.Field;import java.math.BigDecimal;import java.math.RoundingMode;import java.util.concurrent.TimeUnit;import com.google.common.base.Stopwatch;import com.google.common.util.concurrent.RateLimiter;public class RateLimiterTest { public static void main(String args[]) throws Exception { RateLimiterWrapper limiter = new RateLimiterWrapper( RateLimiter.create(2)); limiter.acquire(); limiter.acquire(); limiter.acquire(); limiter.acquire(); System.out.println("Thread.sleep 2s"); try { Thread.sleep(2000L); } catch (InterruptedException e) { e.printStackTrace(); } limiter.acquire(); limiter.acquire(); limiter.acquire(); limiter.acquire(); } static class RateLimiterWrapper { private RateLimiter limiter; private Stopwatch stopwatch; private String simpleName; private Field storedPermits; private Field nextFreeTicketMicros; public RateLimiterWrapper(RateLimiter limiter) throws Exception { this.limiter = limiter; Class clas = limiter.getClass(); simpleName = clas.getSimpleName(); storedPermits = clas.getSuperclass().getDeclaredField( "storedPermits"); storedPermits.setAccessible(true); nextFreeTicketMicros = clas.getSuperclass().getDeclaredField( "nextFreeTicketMicros"); nextFreeTicketMicros.setAccessible(true); Field stopwatchField = clas.getSuperclass().getSuperclass() .getDeclaredField("stopwatch"); stopwatchField.setAccessible(true); Object sleepingStopwatch = stopwatchField.get(limiter); Field stopwatchFiled = (sleepingStopwatch.getClass() .getDeclaredField("stopwatch")); stopwatchFiled.setAccessible(true); stopwatch = (Stopwatch) stopwatchFiled.get(sleepingStopwatch); System.out .println(String .format("%s -> 初始化阶段: init-storedPermits: %s, init-nextFreeTicketMicros: %s", simpleName, storedPermits.get(limiter), nextFreeTicketMicros.get(limiter))); } long readMicros() { return stopwatch.elapsed(TimeUnit.MICROSECONDS); } double acquire() throws Exception { long reqTimeMirco = readMicros(); Object beforeStoredPermits = storedPermits.get(this.limiter); Object beforeNextFreeTicketMicros = nextFreeTicketMicros .get(this.limiter); double waitMirco = this.limiter.acquire(); Object afterStoredPermits = storedPermits.get(this.limiter); Object afterNextFreeTicketMicros = nextFreeTicketMicros .get(this.limiter); System.out .println(String .format("reqTimeMirco: %s, before-storedPermits: %s, before-nextFreeTicketMicros: %s, waitSeconds: %ss, after-storedPermits: %s, after-nextFreeTicketMicros: %s", reqTimeMirco, convert(beforeStoredPermits), beforeNextFreeTicketMicros, convert(waitMirco), convert(afterStoredPermits), afterNextFreeTicketMicros)); return waitMirco; } } public static BigDecimal convert(Object o) { return new BigDecimal(String.valueOf(o)).setScale(4, RoundingMode.HALF_UP); }}
执行结果:
此处可以看到设置的桶容量为2(即允许的突发量),这是因为SmoothBursty中有一个参数:最大突发秒数(maxBurstSeconds)默认值是1s,突发量(桶容量)=速率*maxBurstSeconds,所以本示例 桶容量(突发量)为2。我们发现: 在线程等待后,是第四个请求才开始有等待!
处理过程:
- 在resync方法中: 因为 请求时间reqTimeMirco > before-nextFreeTicketMicros, 所以计算得出:
- 截止到此次请求时间点令牌总数 = 原剩余令牌数 + 能够生成的令牌数 (reqTimeMirco - before-nextFreeTicketMicros) / 单个令牌生成时间stableIntervalMicros ), 不超过maxPermits。
- nextFreeTicketMicros 被替换成了reqTimeMirco。
- 在reserveEarliestAvailable方法中: 接着更新 nextFreeTicketMicros = before-nextFreeTicketMicros + 差额令牌数 ( 需求量 - 令牌总数) * 单个令牌生成时间stableIntervalMicro
所以按这个逻辑:
此时的stableIntervalMicros = 500000;-
第一次请求中:
-
虽然 before-nextFreeTicketMicros = 745,但在reserveEarliestAvailable方法中返回的 nextFreeTicketMicros 的值就是reqTimeMirco值19298 ,所以与请求时间比较获得的等待时间waitSeconds = 0s。
-
差额令牌时间 = (需求量 1 - (请求时间 19298 - 原令牌可供时间 745)/ 单个令牌生成时间 500000) * 单个令牌生成时间 500000 = 481447; 所以更新之后的nextFreeTicketMicros = 481447+ 19298 =500745 !
-
-
第二次请求:
-
reqTimeMirco < before-nextFreeTicketMicros,所以不计算能够生成令牌数量。 直接比较等待时间 waitSeconds = 500745 - 21164= 479581= 0.4796s;
-
更新之后的nextFreeTicketMicros = 500745 + 1 * 500000 = 1000745 !
-
SmoothWarmingUp 速率平滑
因为SmoothBursty允许一定程度的突发,会有人担心如果允许这种突发,假设突然间来了很大的流量,那么系统很可能扛不住这种突发。因此需要一种平滑速率的限流工具,从而系统冷启动后慢慢的趋于平均固定速率(即刚开始速率小一些,然后慢慢趋于我们设置的固定速率)
public static void main(String args[]) throws Exception { RateLimiterWrapper limiter = new RateLimiterWrapper(RateLimiter.create( 5, 1, TimeUnit.SECONDS)); limiter.acquire(); limiter.acquire(); limiter.acquire(); limiter.acquire(); limiter.acquire(); limiter.acquire(); limiter.acquire(); }
执行结果:
RateLimiterWrapper limiter = new RateLimiterWrapper(RateLimiter.create(5, 3, TimeUnit.SECONDS));
执行结果:
常用限流算法
漏桶算法
思路: 水(请求)先进入到漏桶里,漏桶以一定的速度出水,当水流入速度过大会直接溢出,可以看出漏桶算法能强行限制数据的传输速率。(类似于队列,控制出队速率)
对于很多应用场景来说,除了要求能够限制数据的平均传输速率外,还要求允许某种程度的突发传输。这时候漏桶算法可能就不合适了,令牌桶算法更为适合。
令牌桶算法
系统会以一个恒定的速度往桶里放入令牌,而如果请求需要被处理,则需要先从桶里获取一个令牌,当桶里没有令牌可取时,则拒绝服务。