限流算法学习分享
最近由于自己的项目需要用到限流算法，所以就自己学习了一下限流算法相关的知识，算法学习部分java代码是自己写的，非常烂，我不保证它是对的，就这样子
1. 限流算法学习：
单个时间内允许部分操作
问题：可能会有流量突刺
举例：如果一个小时只允许10个用户操作 那么在第一个小时的后五分钟和第二个小时的前五分钟中各自出现了10个用户操作，相当于在10分钟内执行了20个用户操作，仍然会有服务器高峰风险
package org.example.types; //固定窗口限流 public class FixedWindow { int size = 0; //限流数 int counter = 0; //计数器 long lastRequestTime = 0L; //当此限流开始时间 long time = 0; //限流时间大小 public FixedWindow(int counter,long time) { this.size = counter; this.counter = counter; this.time = time; } public boolean check() { if(System.currentTimeMillis() - lastRequestTime > time * 1000 || lastRequestTime == 0L) { counter = size - 1; this.lastRequestTime = System.currentTimeMillis(); return true; } else { this.counter = this.counter - 1; return counter >= 0; } } }
单位时间内允许部分操作，但是这个单位时间是滑动的，需要指定一个滑动单位
将一个时间段切分为了几个更小的切片，随着时间的前进，滑动窗口会不断地向前加载切片，向后丢弃切片，但是切片的总长度是固定的，滑动窗口保证了自己时间段内所有的访问不会超过阈值
优点：可以解决流量突刺问题
缺点：实现相对复杂，限流效果和滑动单位有关，滑动单位越小，限流效果越好 (理解一下：如果按照上一个的例子，将滑动单位选为一个小时，那么还是会出现固定窗口限流的问题)
package org.example.types; import java.util.HashMap; import java.util.Iterator; import java.util.LinkedHashMap; import java.util.Map; //滑动窗口限流 public class SlideWindow { int size = 0; //限流数 LinkedHashMap<Long,Integer> linkedHashMap = new LinkedHashMap<>(); int windows_num = 0; int windows_size = 0; long start_time = 0; public SlideWindow(int size,int windows_num,int windows_size,long start_time) { this.size = size; this.windows_num = windows_num; this.windows_size = windows_size * 1000; this.start_time = start_time; } public boolean check(){ int count = 0; long l = (System.currentTimeMillis() - this.start_time)/windows_size; long window_time = this.start_time + l*windows_size; if(window_time - windows_num * windows_size - this.start_time >= 0){ this.start_time = window_time - windows_num * (windows_size -1); } Iterator<Map.Entry<Long, Integer>> iterator = linkedHashMap.entrySet().iterator(); while (iterator.hasNext()){ Map.Entry<Long, Integer> entry = iterator.next(); if(entry.getKey() < this.start_time){ iterator.remove(); } else { count++; } } if(count >= this.size){ System.out.println(linkedHashMap); return false; } linkedHashMap.put(window_time,linkedHashMap.containsKey(window_time)?linkedHashMap.get(window_time)+1:1); System.out.println(linkedHashMap); return true; } }
以固定的速率处理请求，当请求桶满了后，拒绝请求
每秒处理10个请求，桶的容量是10，每0.1秒固定处理一次请求，如果1秒来了10个请求，都可以处理完，但如果1秒内来了11个请求，请求就会溢出桶，会被拒绝
优点：能够一定程度上应对流量突刺，能以固定速率处理请求，保证服务器的安全
缺点：没有办法迅速处理一批请求，只能一个一个按顺序来处理
package org.example.types; public class LeakyBucket { private long rate; private long water = 0; private long max = 0; private long last_request = 0; public LeakyBucket(long rate, long max) { this.rate = rate; this.max = max; } public boolean check(){ System.out.println("last_request = " + last_request); if(last_request == 0){ last_request = System.currentTimeMillis(); water = 1; System.out.println("water_now "+water); return true; } long now = System.currentTimeMillis(); this.water =(this.water - (now - last_request)/1000 * rate) >= 0 ? this.water - (now - last_request)/1000 * rate : 0; if((now - last_request)/1000 != 0){ this.last_request = this.last_request+1000 * ((int) (now - last_request)/1000); } if(this.water + 1 > max){ System.out.println("water_now "+water); return false; } else { this.water = this.water + 1; System.out.println("water_now "+water); return true; } } }
管理员首先生成一批令牌，每秒生成10个令牌，用户操作前会先去拿到一个令牌，有令牌的人就有资格执行操作，能同时执行操作，拿不到令牌就等待
优点：能够并发处理同时的请求，并发性能更高
需要考虑的问题：存在时间选取的问题
package org.example.types; public class MakeBucket { private long refreshTime; private long capacity; private long currentToken = 0L; boolean tokenBucketTryAcquire() { long currentTime = System.currentTimeMillis(); long generateToken = (currentTime - refreshTime) / 1000 * putTokenRate; currentToken = Math.min(capacity, generateToken + currentToken); refreshTime = currentTime; // 刷新时间 if (currentToken > 0) { currentToken--; return true; } return false; } }
package org.example; import org.example.types.FixedWindow; import org.example.types.LeakyBucket; import org.example.types.MakeBucket; import org.example.types.SlideWindow; public class Main { public static void main(String[] args) throws InterruptedException { testSlideWindow(); } public static void testFixedWindow() throws InterruptedException { FixedWindow fixedWindow = new FixedWindow(10,10); for (int i=0 ; i<=11 ; i++){ if(i==11){ Thread.sleep(10000); } boolean check = fixedWindow.check(); System.out.println(check); } } public static void testSlideWindow() throws InterruptedException { SlideWindow slideWindow = new SlideWindow(4,5,1,System.currentTimeMillis()); for (int i=0;i<=5;i++){ boolean check = slideWindow.check(); System.out.println(check); Thread.sleep(1000); if(i==3){ Thread.sleep(4000); } } } public static void testLeakyBucket() throws InterruptedException { LeakyBucket leakyBucket = new LeakyBucket(1,5); for (int i=0;i<=10;i++){ boolean check = leakyBucket.check(); System.out.println(check); Thread.sleep(500); if(i==5){ Thread.sleep(3000); } } } public static void testMakeBucket() throws InterruptedException { MakeBucket makeBucket = new MakeBucket(10,1); for (int i=0;i<=10;i++){ boolean check = makeBucket.check(); System.out.println(check); Thread.sleep(500); if(i==9){ Thread.sleep(10000); } } } }
2. 组件学习：
2.1. Guava源码学习：
2.1.1. 算法：
令牌桶算法
2.1.2. 源码介绍：
Guava将限流器分为了两种类型：
SmoothBursty - 不带有预热的限流器
SmoothWarmingUp - 带有预热的限流器
由于两种限流器的核心方法基本相同，只主要说明不带有预热器的限流算法：
2.1.2.1. 核心参数说明：
abstract class SmoothRateLimiter extends RateLimiter { // 存储的令牌数量 double storedPermits; // 允许存储的最大令牌数量 double maxPermits; // 添加令牌的间隔时间 double stableIntervalMicros; // 下次刷新令牌的时间需要大于这个时间 private long nextFreeTicketMicros; }
2.1.2.2. 核心函数介绍
final void doSetRate(double permitsPerSecond, long nowMicros) { //生成令牌更新下一期令牌生成时间 this.resync(nowMicros); //更新添加令牌时间间隔 double stableIntervalMicros = (double)TimeUnit.SECONDS.toMicros(1L) / permitsPerSecond; this.stableIntervalMicros = stableIntervalMicros; this.doSetRate(permitsPerSecond, stableIntervalMicros); }
void resync(long nowMicros) { //如果当前时间大于之前记录的令牌刷新时间 if (nowMicros > this.nextFreeTicketMicros) { //计算需要生成令牌数量 double newPermits = (double)(nowMicros - this.nextFreeTicketMicros) / this.coolDownIntervalMicros(); //更新存储的令牌数量 this.storedPermits = Math.min(this.maxPermits, this.storedPermits + newPermits); //更新令牌刷新时间 this.nextFreeTicketMicros = nowMicros; } }
//不加预热的话maxBurstSeconds 默认为1 相对于发了1s的令牌 void doSetRate(double permitsPerSecond, double stableIntervalMicros) { double oldMaxPermits = this.maxPermits; this.maxPermits = this.maxBurstSeconds * permitsPerSecond; if (oldMaxPermits == Double.POSITIVE_INFINITY) { this.storedPermits = this.maxPermits; } else { this.storedPermits = oldMaxPermits == 0.0 ? 0.0 : this.storedPermits * this.maxPermits / oldMaxPermits; } }
public double acquire(int permits) { //计算需要等待的时间 long microsToWait = this.reserve(permits); this.stopwatch.sleepMicrosUninterruptibly(microsToWait); return 1.0 * (double)microsToWait / (double)TimeUnit.SECONDS.toMicros(1L); }
final long reserve(int permits) { checkPermits(permits); synchronized(this.mutex()) { return this.reserveAndGetWaitLength(permits, this.stopwatch.readMicros()); } } final long reserveAndGetWaitLength(int permits, long nowMicros) { long momentAvailable = this.reserveEarliestAvailable(permits, nowMicros); return Math.max(momentAvailable - nowMicros, 0L); } //所以是可以预支令牌的 final long reserveEarliestAvailable(int requiredPermits, long nowMicros) { //生成令牌 + 更新最后一次生成令牌时间 this.resync(nowMicros); long returnValue = this.nextFreeTicketMicros; //更新存储的令牌 和 令牌刷新的时间 double storedPermitsToSpend = Math.min((double)requiredPermits, this.storedPermits); double freshPermits = (double)requiredPermits - storedPermitsToSpend; long waitMicros = this.storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend) + (long)(freshPermits * this.stableIntervalMicros); this.nextFreeTicketMicros = LongMath.saturatedAdd(this.nextFreeTicketMicros, waitMicros); this.storedPermits -= storedPermitsToSpend; return returnValue; }
2.2. Redission源码学习：
2.2.1. 算法：
滑动窗口限流
2.2.2. 源码介绍：
@Override public boolean trySetRate(RateType type, long rate, long rateInterval, RateIntervalUnit unit) { return get(trySetRateAsync(type, rate, rateInterval, unit)); } //设置redis 限流器配置 //rate - 限流量 interval - 时间 type - 限流类型 @Override public RFuture<Boolean> trySetRateAsync(RateType type, long rate, long rateInterval, RateIntervalUnit unit) { return commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN, "redis.call('hsetnx', KEYS[1], 'rate', ARGV[1]);" + "redis.call('hsetnx', KEYS[1], 'interval', ARGV[2]);" + "return redis.call('hsetnx', KEYS[1], 'type', ARGV[3]);", Collections.singletonList(getRawName()), rate, unit.toMillis(rateInterval), type.ordinal()); }
private <T> RFuture<T> tryAcquireAsync(RedisCommand<T> command, Long value) { return commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, command, //获取限流器的配置 "local rate = redis.call('hget', KEYS[1], 'rate');" + "local interval = redis.call('hget', KEYS[1], 'interval');" + "local type = redis.call('hget', KEYS[1], 'type');" + "assert(rate ~= false and interval ~= false and type ~= false, 'RateLimiter is not initialized')" //根据类型获取对应key的值 + "local valueName = KEYS[2];" + "local permitsName = KEYS[4];" + "if type == '1' then " + "valueName = KEYS[3];" + "permitsName = KEYS[5];" + "end;" //判断请求的流量不能超过限定的流量 + "assert(tonumber(rate) >= tonumber(ARGV[1]), 'Requested permits amount could not exceed defined rate'); " //判断目前存在的令牌数 + "local currentValue = redis.call('get', valueName); " //如果存在当前令牌的值 + "if currentValue ~= false then " + "local expiredValues = redis.call('zrangebyscore', permitsName, 0, tonumber(ARGV[2]) - interval); " + "local released = 0; " //释放不在窗口内的值 + "for i, v in ipairs(expiredValues) do " + "local random, permits = struct.unpack('fI', v);" + "released = released + permits;" + "end; " //计算当前窗口剩余的令牌数量 + "if released > 0 then " + "redis.call('zremrangebyscore', permitsName, 0, tonumber(ARGV[2]) - interval); " + "currentValue = tonumber(currentValue) + released; " + "redis.call('set', valueName, currentValue);" + "end;" //判断是否当前窗口已满，如果没满就分发令牌 + "if tonumber(currentValue) < tonumber(ARGV[1]) then " + "local nearest = redis.call('zrangebyscore', permitsName, '(' .. (tonumber(ARGV[2]) - interval), '+inf', 'withscores', 'limit', 0, 1); " + "return tonumber(nearest[2]) - (tonumber(ARGV[2]) - interval);" + "else " + "redis.call('zadd', permitsName, ARGV[2], struct.pack('fI', ARGV[3], ARGV[1])); " + "redis.call('decrby', valueName, ARGV[1]); " + "return nil; " + "end; " + "else " //初始化滑动窗口 + "redis.call('set', valueName, rate); " + "redis.call('zadd', permitsName, ARGV[2], struct.pack('fI', ARGV[3], ARGV[1])); " + "redis.call('decrby', valueName, ARGV[1]); " + "return nil; " + "end;", Arrays.asList(getRawName(), getValueName(), getClientValueName(), getPermitsName(), getClientPermitsName()), value, System.currentTimeMillis(), ThreadLocalRandom.current().nextLong()); }
