介绍

令牌桶算法：
一个存放固定容量令牌的桶，按照固定速率往桶里添加令牌；如下：
a.假设是2r/s，则每500毫秒添加n个令牌；
b.桶中最多存放x个令牌，桶满时，再添加会拒绝；
c.请求去获取令牌，如果令牌足够，允许通过；如果令牌不足，拒绝请求或放入缓冲区等待；

漏桶算法：
漏桶容量固定，按照固定速率流出水滴；如果桶是空的，不用流，如果桶满了，再流入水滴，则拒绝服务。

区别

令牌桶控制的是平均流入速率，速率可高可低，允许有突发请求；漏桶控制的是恒定的流出速率，从而平滑流入速率。

java实现

令牌桶算法：

package com.cvnavi.oa.report;

import org.apache.kafka.common.protocol.types.Field;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class TokenLimiter {

    private ArrayBlockingQueue<String> blockingQueue;

    //容量大小
    private int limit;

    //令牌的产生间隔
    private int period;

    //令牌每次产生的个数
    private int amount;

    public TokenLimiter(int limit, int period, int amount) {
        this.limit = limit;
        this.period = period;
        this.amount = amount;

        blockingQueue = new ArrayBlockingQueue<>(limit);
    }

    private void init() {
        for(int i = 0; i < limit; i++) {
            blockingQueue.add("lp");
        }
    }

    private void addToken(int amount) {
        for(int i = 0; i < amount; i++) {
            //溢出返回false
            blockingQueue.offer("lp");
        }
    }

    /**
     * 获取令牌
     * @return
     */
    public boolean tryAcquire() {
        //队首元素出队
        return blockingQueue.poll() != null ? true : false;
    }

    /**
     * 生产令牌
     */
    private void start(Object lock) {
        Executors.newScheduledThreadPool(1).scheduleAtFixedRate(() -> {
            synchronized (lock) {
                addToken(this.amount);
                lock.notify();
            }
        }, 500, this.period, TimeUnit.MILLISECONDS);
    }

    /**
     * 先生产2个令牌，减少4个令牌；再每500ms生产2个令牌，减少4个令牌
     */
    public static void main(String[] args) throws InterruptedException {
        int period = 500;
        TokenLimiter limiter = new TokenLimiter(8, period, 2);
        limiter.init();

        Object lock = new Object();
        limiter.start(lock);

        //让线程先产生2个令牌(溢出)
        synchronized (lock) {
            lock.wait();
        }
        for(int i = 0; i < 8; i++) {
            for(int j = 0; j < 4; j++) {
                String s = i + "，" + j + "：";
                if(limiter.tryAcquire()) {
                    System.out.println(s + "拿到令牌");
                }
                else{
                    System.out.println(s + "拒绝");
                }
            }
            Thread.sleep(period);
        }
    }

}

package com.cvnavi.report;

public class Funnel {

    //漏斗容量
    int capacity;
    //漏水速度
    double leakingRate;
    //漏斗剩余空间(不是水的大小)
    int leftQuota;
    //上次漏斗漏水的时间
    long leakingTs;

    public Funnel(int capacity, double leakingRate, int leftQuota, long leakingTs) {
        this.capacity = capacity;
        this.leakingRate = leakingRate;
        this.leftQuota = leftQuota;
        this.leakingTs = leakingTs;
    }

    private void makeSpace() {
        long nowTs = System.currentTimeMillis();
        //这个是间隔的时间
        long deltaTs = nowTs - this.leakingTs;
        //漏掉的水
        int deltaQuota = (int)(deltaTs * leakingRate);
        //间隔时间太长,溢出
        if (deltaQuota < 0){
            this.leftQuota = capacity;
            this.leakingTs = nowTs;
            return;
        }
        //说明漏的时间不够
        if (deltaQuota < 1) {
            return;
        }
        this.leakingTs = nowTs;
        this.leftQuota = this.leftQuota + deltaQuota;
        if (this.leftQuota > this.capacity) {
            this.leftQuota = this.capacity;
        }
    }

    /**
     * 流入
     * @param quota
     * @return
     */
    public Boolean water(int quota) {
        makeSpace();
        //表示量充足
        if (leftQuota >= quota){
            leftQuota =  leftQuota - quota;
            return true;
        }
        //剩余量不够
        return false;
    }

}


package com.cvnavi.report;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

public class FunnelLimitRate {

    private static Map<String,Funnel> funnels = new HashMap<>();


    public static void main(String[] args) throws InterruptedException {
        String userId= "2019";
        String actionKey = "rgister";
        int capacity = 8;
        double leakingRate = 0.001;
        int leftQuota = 5;
        Funnel funnel = funnels.get(userId);
        if (funnel == null) {
            funnel = new Funnel(capacity, leakingRate, leftQuota, System.currentTimeMillis());
        }
        //每1000ms(1s)，流入1的水
        //每1000ms，流出1000*leakingRate(而且要>1)的水
        for (int i = 0; i < 8; i++) {
            Boolean isBoolean = funnel.water(1);
            TimeUnit.MILLISECONDS.sleep(1000);
            System.out.println(isBoolean + " " + funnel.leftQuota);
        }

    }

}

令牌桶算法简介

令牌桶是指一个限流容器，容器有最大容量，每秒或每100ms产生一个令牌（具体取决于机器每秒处理的请求数），当容量中令牌数量达到最大容量时，令牌数量也不会改变了，只有当有请求过来时，使得令牌数量减少（只有获取到令牌的请求才会执行业务逻辑），才会不断生成令牌，所以令牌桶算法是一种弹性的限流算法

限流完下一步我们可以做什么呢，限流完后所有请求都是获取令牌的，都是我们要执行的，但是如果还有很多请求，这时我们要有一个方式决定怎样执行这些请求，这个方式称为调度，可以看我这篇文章 调度算法

令牌桶算法限流范围：

假设令牌桶最大容量为n，每秒产生r个令牌

1. 平均速率：则随着时间推延，处理请求的平均速率越来越趋近于每秒处理r个请求，说明令牌桶算法可以控制平均速率
2. 瞬时速率：如果在一瞬间有很多请求进来，此时来不及产生令牌，则在一瞬间最多只有n个请求能获取到令牌执行业务逻辑，所以令牌桶算法也可以控制瞬时速率

在这里提下漏桶，漏桶由于出水量固定，所以无法应对突然的流量爆发访问，也就是没有保证瞬时速率的功能，但是可以保证平均速率

单机版实现

1. capacity为当前令牌桶的中令牌数量，timeStamp为上一次请求获取令牌的时间，我们没必要真的实现计时器每秒产生多少令牌放入容器中，只要记住上一次请求到来的时间，和这次请求的差值就知道在这段时间内产生了多少令牌
2. 下面假设100ms产生1个令牌，且令牌桶最大容量为10
3. Thread.sleep是模拟不同请求到来的间隔，改变间隔可以看到不同现象

public class LeakyBucketAlgorithm {
    private int capacity = 10;
    private long timeStamp = System.currentTimeMillis();

    public boolean getToken() {
        if (capacity > 0) {
            capacity--;
            return true;
        }
        long current = System.currentTimeMillis();
        if (current - timeStamp >= 100) {
            if ((current - timeStamp) / 100 >= 2) {
            	// 假设100ms产生一个令牌
                capacity += (int)((current - timeStamp) / 100) - 1;
            }
            timeStamp = current;
            if (capacity > 10) capacity = 10;
            return true;
        }
        return false;
    }

    public static void main(String[] args) throws InterruptedException {
        LeakyBucketAlgorithm leakyBucketAlgorithm = new LeakyBucketAlgorithm();
        while (true) {
//            Thread.sleep(10);
            Thread.sleep(100);
            if (leakyBucketAlgorithm.getToken()) {
                System.out.println("获取令牌成功，可以执行业务逻辑了");
            } else {
                System.out.println("获取令牌失败，请稍后重试");
            }
        }
    }
}

多线程版实现

1. 多线程实现也非常简单，只要在方法加一个同步锁，保证同时只有一个请求能尝试获取令牌就可以了
2. Thread.sleep(1000)是因为有10个线程在抢令牌，而每100ms才产生一个令牌，所以每隔1秒抢一次，可以改变这个值看现象

public class LeakyBucketAlgoritmThread {
    private int capacity = 10;
    private long timeStamp = System.currentTimeMillis();


    public synchronized boolean getToken() {
        if (capacity > 0) {
            capacity--;
            return true;
        }
        long current = System.currentTimeMillis();
        if (current - timeStamp >= 100) {
            if ((current - timeStamp) / 100 >= 2) {
                capacity += (int)((current - timeStamp) / 100) - 1;
            }
            timeStamp = current;
            if (capacity > 10) capacity = 10;
            return true;
        }
        return false;
    }

    public static void main(String[] args) throws InterruptedException {
        LeakyBucketAlgoritmThread leakyBucketAlgorithmThread = new LeakyBucketAlgoritmThread();
        for (int i = 0; i < 10; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    while (true) {
                        try {
//                            Thread.sleep(900);
                            Thread.sleep(1000);
                            if (leakyBucketAlgorithmThread.getToken()) {
                                System.out.println(Thread.currentThread().getName()+" 获取令牌成功，可以执行业务逻辑了");
                            } else {
                                System.out.println(Thread.currentThread().getName()+" 获取令牌失败，请稍后重试");
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }
                }
            }).start();
        }
    }
}