前言
 
限流作为高并发场景下抵挡流量洪峰，保护后端服务不被冲垮的一种有效手段，比如大家熟知的限流组件guawa，springcloud中的Hystrix，以及springcloud-alibaba生态中的Sentinel，甚至是基于网关的限流，比如在nginx中配置限流策略，在gateway中配置限流策略等
 
限流无处不在，既然限流的作用如此强大，那么其底层的实现原理如何呢，说到底，限流的核心是由一系列不同的算法完成，本篇将通过实例来说明下常用的几种限流算法的用法和原理
 
1、计数器算法
 
计数器算法限流是采用简单的计数操作，到一段时间之后自动清零，通俗来说，就是系统允许的最大流量是固定的，每过来一个请求分发一个数量的资源处理请求，一旦某个时间段，这批用于处理请求的资源达到了最大值，后续再过来的请求就直接没法处理了
 
为了模拟计数器的效果，这里我们使用Java中的Semaphore，对Semaphore有过了解的同学应该直到，这个组件可以搭配线程一起使用，可以对并发线程进行处理，
 
可以理解Semaphore就是一个令牌发放的人员，所有过来的请求都必须从Semaphore中拿到一个资源(线程)，拿到资源的请求才能被处理
 
/**
 * 计数器限流算法
 */
public class Counter {

    public static void main(String[] args) {
        final Semaphore semaphore = new Semaphore(3);
        ScheduledExecutorService service = Executors.n

 
 
导语
   在Java开发过程中，经常用到的限流算法有两种，一种是令牌桶算法，一种是漏斗桶算法，那么下面就来分别看看两种算法的Java实现方式。
 
 


 
文章目录
 
限流算法之漏桶算法
漏桶的实现
   
限流算法之令牌桶法
简单的令牌桶算法实现
   
总结
 

 
限流算法之漏桶算法
 
  漏桶算法:首先保证请求先进入到漏桶中，漏桶再通过一定的速度将流量引出，这里需要注意的是当流量太大或者是漏桶已经满了之后会直接溢出，然后就开始拒绝请求，可以看出漏桶算法能强行限制数据传输的速率。如图所示，图来自网络
 
 
漏桶的实现
 
  实现的关键点：
 
1、定义这个桶的最大容量；
2、记录好上次操作完成后的时间以及桶的剩余量，用来记录后续计算当前桶的容量；
3、定义好流速，速率越小，流控越大，流量越小；
4、每次请求，需要先检查桶大小，如果超过最大值，则拒绝，如果没有超过最大值则继续进行操作；
 
public class LeakyBucket {

    // 流出速率
    private double rate;

    // 桶大小
    private double burst;

    // 最后更新时间
    private long refreshTime;

    // 现有量
    private int water;


    public LeakyBucket(double rate,double burst){
        this.rate = rate;
        this.burst = burst;
    }

    /**
     * 用来刷新水量
     */
    private void refreshWater(){
        long now = System.currentTimeMillis();
        water = (int) Math.max(0,water-(now-refreshTime)* rate);
        refreshTime = now;
    }

    public synchronized boolean tryAcquire(){
        refreshWater();
        if (water<burst) {
            water++;
            return true;
        }else {
            return false;
        }
    }


    private static LeakyBucket leakyBucket = new LeakyBucket(1,100);
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 1000; i++) {
            executorService.execute(()->{
                System.out.println(leakyBucket.tryAcquire());
            });
        }
        executorService.shutdown();
    }
}
 
限流算法之令牌桶法
 
  系统通过一定的速率往桶里面添加令牌，处理请求之前，需要先从桶里获取一个令牌，当桶里没有令牌的时候则需要进行返回失败。
   大概描述如下：
 
所有的请求在处理之前都需要拿到一个可用的令牌才会被处理；
获取不到令牌，则请求返回失败；
根据限流大小，设置按照一定的速率往桶里添加令牌；
桶要设置最大放置令牌的限制，当桶满的时候、添加新的令牌就会被丢弃或者拒绝；
 
 如图所示
 
 
简单的令牌桶算法实现
 
  实现关键点
 
1、初始化固定数量的令牌放入令牌桶中
2、初始化和开启一个定时任务，定时往令牌桶添加令牌
3、提供一个获取令牌的方法，获取一个令牌，令牌桶中减一，如果令牌桶为空，则返回失败
 
public class TokenLimiter {

    private ArrayBlockingQueue<String> blockingQueue;
    private int limit;
    private TimeUnit timeUnit;
    private int period;


    public TokenLimiter(int limit,int period,TimeUnit timeUnit){
        this.limit = limit;
        this.timeUnit = timeUnit;
        this.period = period;
        blockingQueue = new ArrayBlockingQueue<>(limit);
        init();
        start();
    }



    /**
     * 初始化令牌操作
     */
    private void init() {
        for (int i = 0; i < limit; i++) {
            blockingQueue.add("1");
        }
    }

    /**
     * 获取令牌为空，返回false
     * @return
     */
    public boolean tryAcquire(){
        return blockingQueue.poll()==null?false:true;
    }

    private void addToken(){
        blockingQueue.offer("1");
    }




    private void start() {
        Executors.newScheduledThreadPool(1).scheduleAtFixedRate(()->{
            addToken();
        },10,period,timeUnit);
    }
}

 
总结
 
  上面通过简单的方式实现了两种限流算法，当然还是有很多的改进的地方，这里只是通过简单的代码对两种算法的核心内容进行演示，当然后续高级算法的演变也是基于核心思想来进行的变化。

 
 
在高并发业务场景下，保护系统时，常用的"三板斧"有："熔断、降级和限流"。今天和大家谈谈常用的限流算法的几种实现方式，这里所说的限流并非是网关层面的限流，而是业务代码中的逻辑限流。
 
限流算法常用的几种实现方式有如下四种：
 
计数器
滑动窗口
漏桶
令牌桶
 
下面会展开说每种算法的实现原理和他们自身的缺陷，方便以后我们在实际应用中能够根据不同的情况选择正确的限流算法。
 
计数器
 
算法思想
 
计数器是一种比较简单粗暴的限流算法，其思想是在固定时间窗口内对请求进行计数，与阀值进行比较判断是否需要限流，一旦到了时间临界点，将计数器清零。
 
 
 
面临的问题
 
计数器算法存在“时间临界点”缺陷。比如每一分钟限制100个请求，可以在00:00:00-00:00:58秒里面都没有请求，在00:00:59瞬间发送100个请求，这个对于计数器算法来是允许的，然后在00:01:00再次发送100个请求，意味着在短短1s内发送了200个请求，如果量更大呢，系统可能会承受不住瞬间流量，导致系统崩溃。（如下图所示）
 
 
 
所以计数器算法实现限流的问题是没有办法应对突发流量，不过它的算法实现起来确实最简单的，下面给出一个用Go代码实现的计数器。
 
代码实现
 
type LimitRate struct {
   rate  int           //阀值
   begin time.Time     //计数开始时间
   cycle time.Duration //计数周期
   count int           //收到的请求数
   lock  sync.Mutex    //锁
}

func (limit *LimitRate) Allow() bool {
   limit.lock.Lock()
   defer limit.lock.Unlock()

   // 判断收到请求数是否达到阀值
   if limit.count == limit.rate-1 {
      now := time.Now()
      // 达到阀值后，判断是否是请求周期内
      if now.Sub(limit.begin) >= limit.cycle {
         limit.Reset(now)
         return true
      }
      return false
   } else {
      limit.count++
      return true
   }
}

func (limit *LimitRate) Set(rate int, cycle time.Duration) {
   limit.rate = rate
   limit.begin = time.Now()
   limit.cycle = cycle
   limit.count = 0
}

func (limit *LimitRate) Reset(begin time.Time) {
   limit.begin = begin
   limit.count = 0
}

 
滑动窗口
 
算法思想
 
滑动窗口算法将一个大的时间窗口分成多个小窗口，每次大窗口向后滑动一个小窗口，并保证大的窗口内流量不会超出最大值，这种实现比固定窗口的流量曲线更加平滑。
 
普通时间窗口有一个问题，比如窗口期内请求的上限是100，假设有100个请求集中在前1s的后100ms，100个请求集中在后1s的前100ms，其实在这200ms内就已经请求超限了，但是由于时间窗每经过1s就会重置计数，就无法识别到这种请求超限。
 
对于滑动时间窗口，我们可以把1ms的时间窗口划分成10个小窗口，或者想象窗口有10个时间插槽time slot, 每个time slot统计某个100ms的请求数量。每经过100ms，有一个新的time slot加入窗口，早于当前时间1s的time slot出窗口。窗口内最多维护10个time slot。
 
 
 
面临的问题
 
滑动窗口算法是固定窗口的一种改进，但从根本上并没有真正解决固定窗口算法的临界突发流量问题
 
代码实现
 
主要就是实现滑动窗口算法，不过滑动窗口算法一般是找出数组中连续k个元素的最大值，这里是已知最大值n (就是请求上限）如果超过最大值就不予通过。
 
可以参考Bilibili开源的kratos框架里circuit breaker用循环列表保存timeSlot对象的实现，他们这个实现的好处是不用频繁的创建和销毁timeslot对象。下面给出一个简单的基本实现：
 
type timeSlot struct {
 timestamp time.Time // 这个timeSlot的时间起点
 count     int       // 落在这个timeSlot内的请求数
}

// 统计整个时间窗口中已经发生的请求次数
func countReq(win []*timeSlot) int {
 var count int
 for _, ts := range win {
  count += ts.count
 }
 return count
}

type SlidingWindowLimiter struct {
 mu           sync.Mutex    // 互斥锁保护其他字段
 SlotDuration time.Duration // time slot的长度
 WinDuration  time.Duration // sliding window的长度
 numSlots     int           // window内最多有多少个slot
 windows      []*timeSlot
 maxReq       int // 大窗口时间内允许的最大请求数
}

func NewSliding(slotDuration time.Duration, winDuration time.Duration, maxReq int) *SlidingWindowLimiter {
 return &SlidingWindowLimiter{
  SlotDuration: slotDuration,
  WinDuration:  winDuration,
  numSlots:     int(winDuration / slotDuration),
  maxReq:       maxReq,
 }
}


func (l *SlidingWindowLimiter) validate() bool {
 l.mu.Lock()
 defer l.mu.Unlock()


 now := time.Now()
 // 已经过期的time slot移出时间窗
 timeoutOffset := -1
 for i, ts := range l.windows {
  if ts.timestamp.Add(l.WinDuration).After(now) {
   break
  }
  timeoutOffset = i
 }
 if timeoutOffset > -1 {
  l.windows = l.windows[timeoutOffset+1:]
 }

 // 判断请求是否超限
 var result bool
 if countReq(l.windows) < l.maxReq {
  result = true
 }

 // 记录这次的请求数
 var lastSlot *timeSlot
 if len(l.windows) > 0 {
  lastSlot = l.windows[len(l.windows)-1]
  if lastSlot.timestamp.Add(l.SlotDuration).Before(now) {
   // 如果当前时间已经超过这个时间插槽的跨度，那么新建一个时间插槽
   lastSlot = &timeSlot{timestamp: now, count: 1}
   l.windows = append(l.windows, lastSlot)
  } else {
   lastSlot.count++
  }
 } else {
  lastSlot = &timeSlot{timestamp: now, count: 1}
  l.windows = append(l.windows, lastSlot)
 }


 return result
}


 
滑动窗口实现起来代码有点多，完整可运行的测试代码可以访问我的GitHub链接 https://github.com/kevinyan815/gocookbook/issues/26 拿到，我把这个链接也放到了阅读原文里，点击即可访问。
 
漏桶
 
算法思想
 
漏桶算法是首先想象有一个木桶，桶的容量是固定的。当有请求到来时先放到木桶中，处理请求的worker以固定的速度从木桶中取出请求进行相应。如果木桶已经满了，直接返回请求频率超限的错误码或者页面。
 
 
 
适用场景
 
漏桶算法是流量最均匀的限流实现方式，一般用于流量“整形”。例如保护数据库的限流，先把对数据库的访问加入到木桶中，worker再以db能够承受的qps从木桶中取出请求，去访问数据库。
 
存在的问题
 
木桶流入请求的速率是不固定的，但是流出的速率是恒定的。这样的话能保护系统资源不被打满，但是面对突发流量时会有大量请求失败，不适合电商抢购和微博出现热点事件等场景的限流。
 
代码实现
 
// 一个固定大小的桶，请求按照固定的速率流出
// 请求数大于桶的容量，则抛弃多余请求

type LeakyBucket struct {
   rate       float64    // 每秒固定流出速率
   capacity   float64    // 桶的容量
   water      float64    // 当前桶中请求量
   lastLeakMs int64      // 桶上次漏水微秒数
   lock       sync.Mutex // 锁
}

func (leaky *LeakyBucket) Allow() bool {
   leaky.lock.Lock()
   defer leaky.lock.Unlock()

   now := time.Now().UnixNano() / 1e6
   // 计算剩余水量,两次执行时间中需要漏掉的水
   leakyWater := leaky.water - (float64(now-leaky.lastLeakMs) * leaky.rate / 1000)
   leaky.water = math.Max(0, leakyWater)
   leaky.lastLeakMs = now
   if leaky.water+1 <= leaky.capacity {
      leaky.water++
      return true
   } else {
      return false
   }
}

func (leaky *LeakyBucket) Set(rate, capacity float64) {
   leaky.rate = rate
   leaky.capacity = capacity
   leaky.water = 0
   leaky.lastLeakMs = time.Now().UnixNano() / 1e6
}

 
令牌桶
 
算法思想
 
令牌桶是反向的"漏桶"，它是以恒定的速度往木桶里加入令牌，木桶满了则不再加入令牌。服务收到请求时尝试从木桶中取出一个令牌，如果能够得到令牌则继续执行后续的业务逻辑。如果没有得到令牌，直接返回访问频率超限的错误码或页面等，不继续执行后续的业务逻辑。
 
特点：由于木桶内只要有令牌，请求就可以被处理，所以令牌桶算法可以支持突发流量。
 
 
 
同时由于往木桶添加令牌的速度是恒定的，且木桶的容量有上限，所以单位时间内处理的请求书也能够得到控制，起到限流的目的。假设加入令牌的速度为 1token/10ms，桶的容量为500，在请求比较的少的时候（小于每10毫秒1个请求）时，木桶可以先"攒"一些令牌（最多500个）。当有突发流量时，一下把木桶内的令牌取空，也就是有500个在并发执行的业务逻辑，之后要等每10ms补充一个新的令牌才能接收一个新的请求。
 
参数设置
 
木桶的容量  - 考虑业务逻辑的资源消耗和机器能承载并发处理多少业务逻辑。
 
生成令牌的速度 - 太慢的话起不到“攒”令牌应对突发流量的效果。
 
适用场景
 
适合电商抢购或者微博出现热点事件这种场景，因为在限流的同时可以应对一定的突发流量。如果采用漏桶那样的均匀速度处理请求的算法，在发生热点时间的时候，会造成大量的用户无法访问，对用户体验的损害比较大。
 
代码实现
 
type TokenBucket struct {
   rate         int64 //固定的token放入速率, r/s
   capacity     int64 //桶的容量
   tokens       int64 //桶中当前token数量
   lastTokenSec int64 //上次向桶中放令牌的时间的时间戳，单位为秒

   lock sync.Mutex
}

func (bucket *TokenBucket) Take() bool {
   bucket.lock.Lock()
   defer bucket.lock.Unlock()

   now := time.Now().Unix()
   bucket.tokens = bucket.tokens + (now-bucket.lastTokenSec)*bucket.rate // 先添加令牌
   if bucket.tokens > bucket.capacity {
      bucket.tokens = bucket.capacity
   }
   bucket.lastTokenSec = now
   if bucket.tokens > 0 {
      // 还有令牌，领取令牌
      bucket.tokens--
      return true
   } else {
      // 没有令牌,则拒绝
      return false
   }
}

func (bucket *TokenBucket) Init(rate, cap int64) {
   bucket.rate = rate
   bucket.capacity = cap
   bucket.tokens = 0
   bucket.lastTokenSec = time.Now().Unix()
}

 
总结
 
这几种常用的限流算法实现方式，相互之间没有所谓的"谁是所谓的更优秀的实现方法"，主要还是看具体用在哪里。这几个限流算法的简单实现和测试代码我都放到GitHub上了，除此之外仓库里还有不少有用的小程序，有兴趣的可以点击阅读原文自取。
 
 

  

看到这里了，如果喜欢我的文章就帮我点个赞吧，我会每周通过技术文章分享我的所学所见和第一手实践经验，感谢你的支持。微信搜索关注公众号「网管叨bi叨」每周教会你一个进阶知识，还有专门写给开发工程师的Kubernetes入门教程。
 
 
- END -
 
关注公众号，每周分享给你一个进阶知识
 
 

 
限流的意义
 
限流一般是指在一个时间窗口内对某些操作请求的数量进行限制，比如一个论坛限制用户每秒钟只能发一个帖子，每秒钟只能回复5个帖子。限流可以保证系统的稳定，限制恶意请求，防止因为流量暴增导致系统瘫痪宕机。
 
常用的限流算法有：滑动窗口、漏斗以及令牌桶。
 
得益于redis的数据结构特点，redis实现滑动窗口限流和漏斗限流的非常的便捷。
 
滑动窗口限流的原理和实现
 
以xx论坛限制用户行为为例子，比如一秒内进行某个操作50次，这种行为应该进行限制。
 
滑动窗口就是记录一个滑动的时间窗口内的操作次数，操作次数超过阈值则进行限流。
 
网上找的图：
 

 
在redis中可以用zset数据结构来实现这个功能：
 
用唯一的id作为zset的key，可以是user_id + action_key ，value是当前操作的时间戳。每次新的操作请求进来时，先判断当前时间窗口内记录的操作次数 count，小于阈值max则允许进行操作，超过阈值则进行限流。同时对时间窗口之外的数据进行清理，节省内存。
 
简单代码实现：
 
public boolean isActionAllowed(String userId, String actionKey, int period, int maxCount) {
 
// 生成唯一的key
 
String key = String.format("hist:%s:%s", userId, actionKey);
 
long nowTs = System.currentTimeMillis();
 
// 使用管道
 
Pipeline pipe = jedis.pipelined();
 
pipe.multi();
 
// 添加当前操作当zset中
 
pipe.zadd(key, nowTs, "" + nowTs);
 
// 整理zset，删除时间窗口外的数据
 
pipe.zremrangeByScore(key, 0, nowTs - period * 1000);
 
Response count = pipe.zcard(key);
 
pipe.expire(key, period + 1);
 
pipe.exec();
 
pipe.close();
 
return count.get() <= maxCount;
 
}
 
复制代码
 
zset中的value没有特殊含义，只是用来保证每次操作都是唯一的能够被zset记录。
 
漏斗限流
 
顾名思义就是用一个漏斗来存储(记录请求)，一边向漏斗里面加请求一边将请求漏出去。漏斗的漏嘴有一个流水速率，表示单位时间内流出的水量(数据量)，当加水的速率(单位时间加进去的请求)小于流水速率时漏斗永远不会满。
 
我们不用时刻记录==漏水==，只需记录上一次漏水的开始时间，当一个请求进来时，我们只需要计算上次漏水的时间到当前时间一共漏出的数据量count，用这上次漏水至今的数据总量减去count，了，来判断漏斗是否溢出
 
漏斗算法的Java实现：
 
public class FunnelRateLimiter {
 
static class Funnel {
 
// 漏斗大小
 
int capacity;
 
// 漏嘴流水速率
 
float leakingRate;
 
// 漏斗剩余容量
 
int leftQuota;
 
// 上一次漏水时间
 
long leakingTs;
 
public Funnel(int capacity, float leakingRate) {
 
this.capacity = capacity;
 
this.leakingRate = leakingRate;
 
this.leftQuota = capacity;
 
// 初始化时指定当前时间为第一次漏水时间
 
this.leakingTs = System.currentTimeMillis();
 
}
 
void makeSpace() {
 
long nowTs = System.currentTimeMillis();
 
long deltaTs = nowTs - leakingTs;
 
// 流水速率 * 时间，计算这段时间流出的数据总量
 
int deltaQuota = (int) (deltaTs * leakingRate);
 
if (deltaQuota < 0) { // 间隔时间太长，整数数字过大溢出
 
this.leftQuota = capacity;
 
this.leakingTs = nowTs;
 
return;
 
}
 
if (deltaQuota < 1) { // 腾出空间太小，最小单位是 1
 
return;
 
}
 
this.leftQuota += deltaQuota;
 
this.leakingTs = nowTs;
 
// 流出的数据总量超过漏斗容量，说明漏斗在某些时间是空的，没有实际数据漏出
 
if (this.leftQuota > this.capacity) {
 
this.leftQuota = this.capacity;
 
}
 
}
 
boolean watering(int quota) {
 
makeSpace();
 
//比较当前加入的数据量和漏斗剩余容量
 
if (this.leftQuota >= quota) {
 
this.leftQuota -= quota;
 
return true;
 
}
 
return false;
 
}
 
}
 
// 用一个HashMap存储漏斗数据
 
private Map funnels = new HashMap<>();
 
public boolean isActionAllowed(String userId, String actionKey, int capacity, float leakingRate){
 
// 唯一key
 
String key = String.format("%s:%s", userId, actionKey);
 
Funnel funnel = funnels.get(key);
 
if (funnel == null) {
 
funnel = new Funnel(capacity, leakingRate);
 
funnels.put(key, funnel);
 
}
 
return funnel.watering(1); // 需要 1 个 quota
 
}
 
}
 
复制代码
 
Redis-Cell
 
Redis中提供的一个限流Redis模块：redis-cell ，该模块使用漏斗算法并提供原子性的限流指令。
 
该redis指令：
 
> cl.throttle key capacity operations times 1
 
key：唯一的key代表该漏斗
 
capacity：漏斗初始容量
 
operations times ： 该时间长度内可以进行的最大操作数量，比如 30 60 表示60s内的操作次数最大是30 。
 
1 ： 是一个可选参数，默认值是1
 
复制代码
 
额外加餐——令牌桶限流算法
 
谷歌开源项目Guava中的RateLimiter使用的限流算法就是令牌桶限流算法。
 
网上找的图
 

 
以恒定的速度生成令牌并将令牌加到一个桶里，如果桶满了将其丢弃
 
每个请求进来时都会到桶里拿一个令牌，如果没拿到(桶里没有令牌)拒绝该请求，若拿到则正常执行。
 
Redis,限流算法，Redis 限流算法，滑动窗口，漏斗限流，谷歌令牌桶限流