目录

1、漏桶算法

2、令牌桶算法

3、两种算法的区别

4、限流工具类RateLimiter

4.1 RateLimiter demo

4.2 主要接口

常用的限流算法有两种：漏桶算法和令牌桶算法。

1、漏桶算法

漏桶算法思路很简单，请求先进入到漏桶里，漏桶以固定的速度出水，也就是处理请求，当水加的过快，则会直接溢出，也就是拒绝请求，可以看出漏桶算法能强行限制数据的传输速率。

但是对于很多场景来说，除了要求能够限制数据的平均传输速率外，还要求允许某种程度的突发传输。这时候漏桶算法可能就不合适了，令牌桶算法更为适合。

2、令牌桶算法

令牌桶算法的原理是系统会以一个恒定的速度往桶里放入令牌，而如果请求需要被处理，则需要先从桶里获取一个令牌，当桶里没有令牌可取时，则拒绝服务。

为了更好的理解令牌桶算法，自己造了一个限流算法的简略代码，方便理解。

创建一个需要执行的任务，使用线程池执行。

public class TaskRunnable implements Runnable {
    private Integer reqCount;//已使用令牌数量
    public TaskRunnable(int reqCount) {
        this.reqCount = reqCount;
    }
    @Override
    public void run() {
        System.out.println(Thread.currentThread().getName() + "已经执行...,目前已使用令牌数：" + reqCount);
    }
}

具体执行逻辑和限流逻辑，//自己造轮子

public class TokenTask {
    //关键：时间控制周期+令牌桶容量
    static volatile long timeStamp = getNowTime();//令牌桶计时开始
    static AtomicInteger reqCount = new AtomicInteger(0);//统计调用数
    static final int maxReqCount = 5;//时间周期内最大请求数
    static final long effectiveDuration = 1000;//时间控制周期(秒)
    /**
     * 限流逻辑
     * @return
     */
    public static boolean passControl() {
        long now = getNowTime();//程序执行时间
        if (now < (timeStamp + effectiveDuration)) {//在时间控制范围内
            reqCount.getAndIncrement();
            return reqCount.get() <= maxReqCount;//当前时间范围内超过最大请求控制数
        } else {
            timeStamp = now;//超时后重置
            reqCount = new AtomicInteger(1);//占用一个令牌
            return true;
        }
    }
    //获取执行时间
    public static long getNowTime() {
        return System.currentTimeMillis();
    }

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 10; i++) {
            //Thread.sleep(200); //用来模拟令牌周期
            if (passControl()) {//放行
                executor.execute(new TaskRunnable(reqCount.get()));
            } else {
                System.out.println("被限流，请稍后访问！");
            }
        }
    }
}

限流效果

 注：本算法只是简单模拟了一下限流流程，不能用于生产。

3、两种算法的区别

漏桶算法输入的时候请求不固定，但都会在漏桶里边先保存起来（小于漏桶的容量），然后输出的时候采用的是恒定的速率执行请求，有点像队列的先进先出，只是队列中的元素出队的时间间隔一致。

//有点类似于使用线程池 new SingleThreadExecutor()-单线程串行，newFixedThreadPool() 固定线程

令牌桶算法跟漏桶算法刚好相反，令牌桶的大小就是接口所能承载的最大访问量，令牌的发放是恒速的，而最终能在某一时间处理的请求数不是恒定的，这取决于单位时间内令牌桶中的令牌数量。

4、限流工具类RateLimiter

Google开源工具包Guava提供了限流工具类RateLimiter，该类基于“令牌桶算法”，非常方便使用。RateLimiter经常用于限制对一些物理资源或者逻辑资源的访问速率。它支持两种获取Permits接口，一种是如果拿不到立刻返回false，一种会阻塞等待一段时间看能不能拿到。

4.1 RateLimiter demo

//多任务执行,但每秒执行不超过2个任务
final RateLimiter rateLimiter = RateLimiter.create(2.0);
void submitTasks(List<Runnable> tasks, Executor executor) {
    for (Runnable task : tasks) {
        rateLimiter.acquire(); //默认获取1个令牌
        executor.execute(task);
    }
}

//以每秒5kb内的速度发送-限制数据包大小
final RateLimiter rateLimiter = RateLimiter.create(5000.0);
void submitPacket(byte[] packet) {
    rateLimiter.acquire(packet.length);//获取指定数量的令牌
    networkService.send(packet);
}

//以非阻塞的形式达到降级
if(limiter.tryAcquire()) { //未请求到limiter则立即返回false
    doSomething();
}else{
    doSomethingElse();
}

4.2 主要接口

RateLimiter其实是一个abstract类，但是它提供了几个static方法用于创建RateLimiter：

/**
* 创建一个稳定输出令牌的RateLimiter，保证了平均每秒不超过permitsPerSecond个请求
* 当请求到来的速度超过了permitsPerSecond，保证每秒只处理permitsPerSecond个请求
* 当这个RateLimiter使用不足(即请求到来速度小于permitsPerSecond)，会囤积最多permitsPerSecond个请求
*/
public static RateLimiter create(double permitsPerSecond);

/**
* 创建一个稳定输出令牌的RateLimiter，保证了平均每秒不超过permitsPerSecond个请求
* 还包含一个热身期(warmup period),热身期内，RateLimiter会平滑的将其释放令牌的速率加大，直到起达到最大速率
* 同样，如果RateLimiter在热身期没有足够的请求(unused),则起速率会逐渐降低到冷却状态
*
* 设计这个的意图是为了满足那种资源提供方需要热身时间，而不是每次访问都能提供稳定速率的服务的情况(比如带缓存服务，需要定期刷新缓存的)
* 参数warmupPeriod和unit决定了其从冷却状态到达最大速率的时间
*/
public static RateLimiter create(double permitsPerSecond, long warmupPeriod, TimeUnit unit);

提供了两个获取令牌的方法，不带参数表示获取一个令牌。如果没有令牌则一直等待，返回等待的时间(单位为秒)，没有被限流则直接返回0.0

public double acquire();//默认获取一个令牌
public double acquire(int permits);//获取指定令牌数

尝试获取令牌，分为待超时时间和不带超时时间两种：

//尝试获取一个令牌,立即返回
public boolean tryAcquire();
public boolean tryAcquire(int permits);
//尝试获取permits个令牌，带超时时间
public boolean tryAcquire(long timeout, TimeUnit unit);
public boolean tryAcquire(int permits, long timeout, TimeUnit unit);

下篇这篇文章一点要看：

https://zhuanlan.zhihu.com/p/60979444

参考阅读：

流量控制算法——漏桶算法和令牌桶算法 - 知乎

序言

此两种算法是服务降级的一种实现.

常用于限制我们的对外服务的QPS,即控制对外服务在单位时间内所能处理的请求数量.保护我们的服务不会被海量请求给崩盘cuiyaonan2000@163.com

漏桶算法

漏桶算法的主要思路为: 

1. 在nginx层与controller层加一层(即漏桶层),
2. 用于接收nginx收到的大批量的请求,接收的请求的速度是没有控制的,但是如果超过了漏桶层的最大容量则直接抛弃该请求.
3. 漏桶层将大批量的请求以特定的速度转发给controller层.(当然漏桶层可以放置在需要的任何位置cuiyaonan2000@163.com)

更直观的理解可以参考百度百科的截图如下:

java实现用例

package nan.yao.cui.others.aboutbuckets;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author 崔耀男
 * @des: TODO
 * @date 2021年6月22日 上午11:10:30
 * @package nan.yao.cui.others.aboutbuckets
 */
public class BucketTest {

    // 桶的容量
    private int capacity = 100;

    // 木桶剩余的水滴的量(初始化的时候的空的桶)
    private AtomicInteger water = new AtomicInteger(0);

    // 水滴的流出的速率,这个可以在 构造方法种设置,比如每秒10个请求.
    private int outRate;

    // 记录上次成功接收到请求的时间
    // 用于计算当前系统时间减去上次请求时间 乘以outrate 所处理的请求数.
    private long receivedTime;

    // 判断该controller是否能继续接收请求
    // true:表示可以处理该请求
    // false:表示不能处理该请求,漏桶已经满了
    public boolean acquire() {

	// 如果是空桶,则直接将
	if (water.get() == 0) {

	    receivedTime = System.currentTimeMillis();
	    water.addAndGet(1);
	    return true;

	}

	// 先计算下上成功接受到当前时间已经流出的记录数
	int outNum = ((int) ((System.currentTimeMillis() - receivedTime) / 1000)) * outRate;
	int waterLeft = water.get() - outNum;
	water.set(Math.max(0, waterLeft));
	// 重新更新leakTimeStamp
	// outNum是否大于0---cuiyaonan2000@163.com
	if (outNum > 0) {
	    receivedTime = System.currentTimeMillis();
	}

	// 尝试加水,并且水还未满
	if ((water.get()) < capacity) {
	    water.addAndGet(1);
	    return true;
	} else {
	    // 水满，拒绝加水
	    return false;
	}
    }

    public static void main(String[] args) {
	System.out.println((((System.currentTimeMillis() - 100) / 1000) * 2));
	System.out.println(System.currentTimeMillis() - 100);
	System.out.println((System.currentTimeMillis() - 100) / 1000);

    }

}

令牌桶算法

令牌桶的算法中心逻辑是: 

1. 以恒定的速度向令牌桶种添加令牌.当令牌桶满时则放弃向令牌桶中添加令牌
2. 请求到达controller层时,需要去令牌桶中获取令牌,如果存在令牌,则继续执行,不存在这放弃这次请求cuiyaonan2000@163.com

更直观的流程图参考百度:

java实现用例

package nan.yao.cui.others.aboutbuckets;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author 崔耀男
 * @des: TODO
 * @date 2021年6月22日 上午11:14:29
 * @package nan.yao.cui.others.aboutbuckets
 */
public class TokenBucketTest {
    
    // 令牌桶的容量----同时表示该controller的同一时间的并发量
    private int capacity = 10;

    // 令牌桶
    private AtomicInteger tokenBucket = new AtomicInteger(0);

    // 每秒钟产生的令牌数量
    private int tokenNum;

    public void doWhile() throws InterruptedException {
	// 多少毫秒产生一个令牌
	int forNum = 1000 / tokenNum;
	while (true) {
	    tokenBucket.set(0);
	    while (forNum < 1000) {
		Thread.sleep(forNum);
		forNum += forNum;
		tokenBucket.getAndAdd(1);
	    }
	}
    }

    public synchronized boolean getToken() {
	if (tokenBucket.get() <= 0) {
	    return false;
	}

	tokenBucket.decrementAndGet();

	return true;
    }

    public static void main(String[] args) {
	System.out.println((((System.currentTimeMillis() - 100) / 1000) * 2));
	System.out.println(System.currentTimeMillis() - 100);
	System.out.println((System.currentTimeMillis() - 100) / 1000);

    }

}

近期在研究Jaeger，Jaeger中有一种采集策略是速率限制类型，内部使用的是漏桶算法，在这里研究了下Jaeger漏桶算法的实现原理，自己仿照其实现了一个rateLimiter，并进行了相关测试，下面是主要实现。

• lck：lck是互斥锁，主要用来防止并发情况下产生错误。
• rate：速率，即接口每秒限制多少个请求。在这里也就是水滴从漏桶中流出的速度，同时也是余量增加的速度。
• balance：漏桶的空闲余量，会随着漏桶滴水逐渐变大；如果将请求添加到漏桶中，会逐渐变小。当请求到来时，如果余量不足1，那么表明不能容下当前的请求，当前的请求会被拒绝。
• limit：漏桶的最大容量。
• lastTime：上次调用Check函数的时间。用于计算时间差dur，然后计算这段时间漏桶流出的水w，增加的余量=流出的水量w=时间*速率=dur*rate。

rateLimiter实现代码：

package ratelimiter

import (
	"sync"
	"time"
)

type rateLimiter struct {
	lck      *sync.Mutex
	rate     float64   //最大速率限制
	balance  float64   //漏桶的余量
	limit    float64   //漏桶的最大容量限制
	lastTime time.Time //上次检查的时间
}

func NewRateLimiter(limitPerSecond int, balance int) *rateLimiter {
	return &rateLimiter{
		lck:      new(sync.Mutex),
		rate:     float64(limitPerSecond),
		balance:  float64(balance),
		limit:    float64(balance),
		lastTime: time.Now(),
	}
}

func (r *rateLimiter) Check() bool {
	ok := false
	r.lck.Lock()
	now := time.Now()
	dur := now.Sub(r.lastTime).Seconds()
	r.lastTime = now
	water := dur * r.rate //计算这段时间内漏桶流出水的流量water
	r.balance += water    //漏桶流出water容量的水，自然漏桶的余量多出water
	if r.balance > r.limit {
		r.balance = r.limit
	}
	if r.balance >= 1 { //漏桶余量足够容下当前的请求
		r.balance -= 1
		ok = true
	}
	r.lck.Unlock()
	return ok
}

单元测试代码：

package ratelimiter

import (
	"fmt"
	"testing"
	"time"
)

func TestRateLimiter_Check(t *testing.T) {
	limiter := NewRateLimiter(10, 1)
	start := time.Now()
	count := 0
	for i := 0; i < 1e3; i++ {
		if limiter.Check() {
			fmt.Println(i)
			count++
		}
		time.Sleep(time.Millisecond)
	}
	fmt.Println("count:", count)
	fmt.Println(time.Now().Sub(start).Seconds())
}

测试结果：每秒限制10个，1.3秒接受了13个请求，加上一开始填满桶的一个，共14个。