常见限流算法研究与实现

柔光的暖阳◎ 2021-09-15 19:28 318阅读 0赞

一、限流场景

很多做服务接口的人或多或少的遇到这样的场景，由于业务应用系统的负载能力有限，为了防止非预期的请求对系统压力过大而拖垮业务应用系统。也就是面对大流量时，如何进行流量控制？服务接口的流量控制策略：分流、降级、限流等。本文讨论下限流策略，虽然降低了服务接口的访问频率和并发量，却换取服务接口和业务应用系统的高可用。

1、实际场景中常用的限流策略：

*  Nginx前端限流

2、按照一定的规则如帐号、IP、系统调用逻辑等在Nginx层面做限流

3、业务应用系统限流

*  客户端限流
 *  服务端限流

4、数据库限流

*  红线区，力保数据库

本文主要讲的是服务端限流的算法。常见的限流算法有：计数器、令牌桶和漏桶算法,下面一起来看看。

二、限流算法

1、计数器

计数器法是限流算法里最简单也是最容易实现的一种算法。比如我们规定，对于A接口来说，我们1分钟的访问次数不能超过10个。那么我们可以这么做：在一开始的时候，我们可以设置一个计数器counter，每当一个请求过来的时候，counter就加1，如果counter的值大于10并且该请求与第一个 请求的间隔时间还在1分钟之内，那么说明请求数过多；如果该请求与第一个请求的间隔时间大于1分钟，且counter的值还在限流范围内，那么就重置 counter，具体算法的示意图如下：

![70][]

（1）写法一，就用一个计算器来实现

public class CountRateLimiter {
    
    /**计算器*/
    private              AtomicLong counter   = new AtomicLong(0);
    /**初始时间*/
    private static       long       timestamp = System.currentTimeMillis();
    /**时间窗口内最大请求个数*/
    private long limit;
    
    public CountRateLimiter(long limit) {
        this.limit = limit;
    }
    
    public boolean tryAcquire() {
        long now = System.currentTimeMillis();
        //1s之内的请求
        if (now - timestamp < 1000) {
            if (counter.get() < limit) {
                counter.incrementAndGet();
                System.out.println("pass_request");
                return true;
            } else {
                System.out.println("refuse_request");
                return false;
            }
        } else {
            counter = new AtomicLong(0);
            timestamp = now;
            System.out.println("time_end,refuse_request");
            return false;
        }
    }
    
    public static void main(String[] args) {
        CountRateLimiter rateLimiter = new CountRateLimiter(10);
        ExecutorService executor = Executors.newCachedThreadPool();
        for(int i=0;i<1000;i++) {
            executor.submit(() -> {
                double random = (new Random()).nextDouble();
                long a = (long)(random * 1000);
                try {
                    //睡眠一下
                    Thread.sleep(a);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                rateLimiter.tryAcquire();
            });
        }
        executor.shutdown();
    }
    }

输出结果：

![70 1][]

（2）写法二，用缓存来实现

使用Guava的Cache来存储计数器，过期时间设置为2秒（保证1秒内的计数器是有的），然后我们获取当前时间戳然后取秒数来作为KEY进行计数统计和限流。

public class GuavaCountRateLimiter {
    
    /**缓存*/
    private LoadingCache<Long, AtomicLong> counter =
            CacheBuilder.newBuilder()
                    .expireAfterWrite(2, TimeUnit.SECONDS)
                    .build(new CacheLoader<Long, AtomicLong>() {
                        @Override
                        public AtomicLong load(Long seconds) throws Exception {
                            return new AtomicLong(0);
                        }
                    });
    
    /**限制每秒10*/
    private long limit = 10;
    
    /**
     * 使用Guava的Cache来存储计数器，过期时间设置为2秒（保证1秒内的计数器是有的），然后我们获取当前时间戳然后取秒数来作为KEY进行计数统计和限流
     */
    public boolean tryAcquire() throws ExecutionException {
        //得到当前秒
        long currentSeconds = System.currentTimeMillis() / 1000;
        if (counter.get(currentSeconds).incrementAndGet() > limit) {
            System.out.println("refuse_request:count="+ counter.get(currentSeconds));
            return true;
        } else {
            System.out.println("pass_request:count=" + counter.get(currentSeconds));
            return false;
        }
    }
    public static void main(String[] args) {
        GuavaCountRateLimiter guavaCountRateLimiter = new GuavaCountRateLimiter();
        ExecutorService executor = Executors.newCachedThreadPool();
        for(int i=0;i<1000;i++) {
            executor.submit(() -> {
                double random = (new Random()).nextDouble();
                long a = (long)(random * 1000);
                try {
                    //睡眠一下
                    Thread.sleep(a);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                try {
                    guavaCountRateLimiter.tryAcquire();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            });
        }
        executor.shutdown();
    }
    }

输出结果：

![70 2][]

2、漏桶算法

漏桶算法即leaky bucket是一种非常常用的限流算法，可以用来实现流量整形（Traffic Shaping）和流量控制（Traffic Policing）。贴了一张维基百科上示意图帮助大家理解：

![70 3][]

漏桶算法的主要概念如下：

*  一个固定容量的漏桶，按照常量固定速率流出水滴；
 *  如果桶是空的，则不需流出水滴；
 *  可以以任意速率流入水滴到漏桶；
 *  如果流入水滴超出了桶的容量，则流入的水滴溢出了（被丢弃），而漏桶容量是不变的。

代码实现

public class LeakRateLimit {
    //定义桶的大小
    private final ConcurrentLinkedQueue<Integer> container = new ConcurrentLinkedQueue<>();
    
    private final static int BUCKET_LIMIT = 1000;
    
    //消费者 不论多少个线程，每秒最大的处理能力是1秒中执行10次
    private final RateLimiter consumerRate = RateLimiter.create(10d);
    
    //往桶里面放数据时，确认没有超过桶的最大的容量
    private Monitor offerMonitor = new Monitor();
    
    //从桶里消费数据时，桶里必须存在数据
    private Monitor consumerMonitor = new Monitor();
    
    /**
     * 往桶里面写数据
     * @param data
     */
    public void submit(Integer data) {
        if (offerMonitor.enterIf(offerMonitor.newGuard(() -> container.size() < BUCKET_LIMIT))) {
            try {
                container.offer(data);
                System.out.println(currentThread() + " submit.." + data + " container size is :[" + container.size() + "]");
            } finally {
                offerMonitor.leave();
            }
        } else {
            //这里时候采用降级策略了。消费速度跟不上产生速度时，而且桶满了，抛出异常
            //或者存入MQ DB等后续处理
            System.out.println("The bucket is ful..Pls latter can try...");
            throw new IllegalStateException(currentThread().getName() + "The bucket is ful..Pls latter can try...");
        }
    }
    
    /**
     * 从桶里面消费数据
     * @param consumer
     */
    public void takeThenConsumer(Consumer<Integer> consumer) {
        if (consumerMonitor.enterIf(consumerMonitor.newGuard(() -> !container.isEmpty()))) {
            try {
                //不打印时 写 consumerRate.acquire();
                System.out.println(currentThread() + "  waiting" + consumerRate.acquire());
                Integer data = container.poll();
                //container.peek() 只是去取出来不会删掉
                consumer.accept(data);
            } finally {
                consumerMonitor.leave();
            }
        } else {
            //当木桶的消费完后，可以消费那些降级存入MQ或者DB里面的数据
            System.out.println("will consumer Data from MQ...");
            try {
                TimeUnit.SECONDS.sleep(10);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    
    public static void main(String[] args) {
        final LeakRateLimit bucket = new LeakRateLimit();
        final AtomicInteger DATA_CREATOR = new AtomicInteger(0);
    
        //生产线程 10个线程 每秒提交 50个数据  1/0.2s*10=50个
        IntStream.range(0, 10).forEach(i -> {
            new Thread(() -> {
                for (; ; ) {
                    int data = DATA_CREATOR.incrementAndGet();
                    try {
                        bucket.submit(data);
                        TimeUnit.MILLISECONDS.sleep(200);
                    } catch (Exception e) {
                        //对submit时，如果桶满了可能会抛出异常
                        if (e instanceof IllegalStateException) {
                            System.out.println(e.getMessage());
                            //当满了后，生产线程就休眠1分钟
                            try {
                                TimeUnit.SECONDS.sleep(60);
                            } catch (InterruptedException e1) {
                                e1.printStackTrace();
                            }
                        }
                    }
                }
            }).start();
        });
    
        //消费线程  采用RateLimiter每秒处理10个  综合的比率是5:1
        IntStream.range(0, 10).forEach(i -> {
            new Thread(
                    () -> {
                        for (; ; ) {
                            bucket.takeThenConsumer(x -> {
                                System.out.println(currentThread() + "C.." + x);
                            });
                        }
                    }
            ).start();
        });
    
    }
    }

输出结果：

![70 4][]

漏桶算法比较好实现，在单机系统中可以使用队列来实现，在分布式环境中消息中间件或者Redis都是可选的方案。

3、令牌桶算法

![70 5][]

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

令牌桶算法是一个存放固定容量令牌（token）的桶，按照固定速率往桶里添加令牌。令牌桶算法基本可以用下面的几个概念来描述：

*  令牌将按照固定的速率被放入令牌桶中。比如每秒放10个。
 *  桶中最多存放b个令牌，当桶满时，新添加的令牌被丢弃或拒绝。
 *  当一个n个字节大小的数据包到达，将从桶中删除n个令牌，接着数据包被发送到网络上。
 *  如果桶中的令牌不足n个，则不会删除令牌，且该数据包将被限流（要么丢弃，要么缓冲区等待）。

代码实现：

public class TokenLimiter {
    
    /**默认桶大小个数 即最大瞬间流量是64M*/
    private static final int DEFAULT_BUCKET_SIZE = 1024 * 1024 * 64;
    
    /**一个桶的单位是1字节*/
    private int everyTokenSize = 1;
    
    /**令牌最大数目*/
    private int bucketSize;
    
    /**平均流量*/
    private int avgFlowRate;
    
    /**队列来缓存桶数量*/
    private ArrayBlockingQueue<Byte> tokenQueue = new ArrayBlockingQueue<Byte>(
            DEFAULT_BUCKET_SIZE);
    
    private ScheduledExecutorService scheduledExecutorService = Executors
            .newSingleThreadScheduledExecutor();
    
    private volatile boolean isStart = false;
    
    private ReentrantLock lock = new ReentrantLock(true);
    
    private static final byte A_CHAR = 'a';
    
    public TokenLimiter() {
    }
    
    public TokenLimiter(int bucketSize, int avgFlowRate) {
        this.bucketSize = bucketSize;
        this.avgFlowRate = avgFlowRate;
    }
    
    public TokenLimiter(int everyTokenSize, int bucketSize, int avgFlowRate) {
        this.everyTokenSize = everyTokenSize;
        this.bucketSize = bucketSize;
        this.avgFlowRate = avgFlowRate;
    }
    
    public void addTokens(Integer tokenNum) {
    
        // 若是桶已经满了，就不再家如新的令牌
        for (int i = 0; i < tokenNum; i++) {
            tokenQueue.offer(Byte.valueOf(A_CHAR));
        }
    }
    
    public TokenLimiter build() {
    
        start();
        return this;
    }
    
    /**
     * 获取足够的令牌个数
     *
     * @return
     */
    public boolean getTokens(byte[] dataSize) {
        //传输内容大小对应的桶个数
        int needTokenNum = dataSize.length / everyTokenSize + 1;
    
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            //是否存在足够的桶数量
            boolean result = needTokenNum <= tokenQueue.size();
            if (!result) {
                return false;
            }
    
            int tokenCount = 0;
            for (int i = 0; i < needTokenNum; i++) {
                Byte poll = tokenQueue.poll();
                if (poll != null) {
                    tokenCount++;
                }
            }
    
            return tokenCount == needTokenNum;
        } finally {
            lock.unlock();
        }
    }
    
    public void start() {
    
        // 初始化桶队列大小
        if (bucketSize > 0) {
            tokenQueue = new ArrayBlockingQueue<Byte>(bucketSize);
        }
    
        // 初始化令牌生产者
        TokenProducer tokenProducer = new TokenProducer(avgFlowRate, this);
        //定时1s生产令牌
        scheduledExecutorService.scheduleAtFixedRate(tokenProducer, 0, 1,
                TimeUnit.SECONDS);
        isStart = true;
    
    }
    
    public void stop() {
        isStart = false;
        scheduledExecutorService.shutdown();
    }
    
    public boolean isStarted() {
        return isStart;
    }
    
    class TokenProducer implements Runnable {
    
        private int          avgFlowRate;
        private TokenLimiter tokenLimiter;
    
        public TokenProducer(int avgFlowRate, TokenLimiter tokenLimiter) {
            this.avgFlowRate = avgFlowRate;
            this.tokenLimiter = tokenLimiter;
        }
    
        @Override
        public void run() {
            tokenLimiter.addTokens(avgFlowRate);
        }
    }
    
    public static TokenLimiter newBuilder() {
        return new TokenLimiter();
    }
    
    public TokenLimiter everyTokenSize(int everyTokenSize) {
        this.everyTokenSize = everyTokenSize;
        return this;
    }
    
    public TokenLimiter bucketSize(int bucketSize) {
        this.bucketSize = bucketSize;
        return this;
    }
    
    public TokenLimiter avgFlowRate(int avgFlowRate) {
        this.avgFlowRate = avgFlowRate;
        return this;
    }
    
    private String stringCopy(String data, int copyNum) {
    
        StringBuilder sbuilder = new StringBuilder(data.length() * copyNum);
    
        for (int i = 0; i < copyNum; i++) {
            sbuilder.append(data);
        }
    
        return sbuilder.toString();
    
    }
    
    public static void main(String[] args) throws IOException,
            InterruptedException {
        TokenLimiter tokenLimiter = TokenLimiter.newBuilder().avgFlowRate(512)
                .bucketSize(1024).build();
    
        BufferedWriter bufferedWriter = new BufferedWriter(
                new OutputStreamWriter(new FileOutputStream("D:/ds_test")));
        String data = "xxxx";// 四个字节
    
        ExecutorService executor = Executors.newCachedThreadPool();
        //初始化
        IntStream.range(0, 1000).forEach(i -> {
            executor.submit(() -> {
                //每个线程需要一个令牌
                boolean token = tokenLimiter.getTokens("x".getBytes());
                if (token) {
                    System.out.println("token pass");
                } else {
                    System.out.println("token refuse");
                }
            });
        });
    }
    
    }

输出结果：

![70 6][]

google开源工具包guava提供了限流工具类RateLimiter，该类基于“令牌桶算法”，非常方便使用

使用方法示例：

public class RateLimiterTest {
    
    public static void main(String[] args) {
        //每秒生产两个令牌
        final RateLimiter rateLimiter = RateLimiter.create(20.0);
    
        ExecutorService executorService = Executors.newCachedThreadPool();
        IntStream.range(0, 10).forEach(i -> {
            executorService.submit(() -> {
                //随机休眠
                Random random = new Random();
                int r = random.nextInt(1000);
                try {
                    TimeUnit.MICROSECONDS.sleep(r);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                //每个线程需要一个令牌
                boolean token = rateLimiter.tryAcquire();
                if (token) {
                    System.out.println("token pass");
                } else {
                    System.out.println("token refuse");
                }
            });
        });
    
    }
    }

输出结果：

![70 7][]

4、漏桶和令牌桶比较

令牌桶可以在运行时控制和调整数据处理的速率，处理某时的突发流量。放令牌的频率增加可以提升整体数据处理的速度，而通过每次获取令牌的个数增加或者放慢令牌的发放速度和降低整体数据处理速度。而漏桶不行，因为它的流出速率是固定的，程序处理速度也是固定的。整体而言，令牌桶算法更优，但是实现更为复杂一些。

[70]: /images/20210811/2edd8288b7db46fc9fae44fcbbd066d9.png
[70 1]: /images/20210811/c1ea6c3516dd49ac831cf33f0a53643c.png
[70 2]: /images/20210811/154b08a23b7f4e148105f4d60d778105.png
[70 3]: /images/20210811/9bcb89e02b1146458876b76c1acf9bde.png
[70 4]: /images/20210811/517e79ae1d94431d86287927297ab4e4.png
[70 5]: /images/20210811/393f6a7f74ac4ac0a07ad981790cef2f.png
[70 6]: /images/20210811/1cb4e6fe487b418db88193aab5060296.png
[70 7]: /images/20210811/906c63bb68e54412acb85e6d361bc497.png