限流算法(令牌桶漏桶计数器)

news2024/11/28 1:42:57

📝个人主页：五敷有你
🔥系列专栏：Spring
⛺️稳中求进，晒太阳

业务重的三种情况：突发流量、恶意流量、业务本身需要

限流: 是为了保护自身系统和下游系统不被高并发流量冲垮，导致系统雪崩。

保证系统在可用的情况下尽可能增加进入的请求，其余的请求在排队等待，或者返回友好提示。保证进入系统的用户可以友好使用。

令牌桶算法

令牌桶算法是一个设定的速率产生令牌(token) 并放入令牌通，每次用户请求都得申请令牌。如果令牌不足，则拒绝请求。

令牌桶算法中新请求到来时会从桶中拿走一个令牌，如果桶内没有i令牌可拿，就拒绝服务。

当然令牌的数量也是有上限的。令牌的数量与时间和发放速率强相关。时间流逝的时间越长，会不断往桶里加入越多的令牌，如果令牌发送的速度比申请速度快，令牌会放满令牌桶，直到令牌占满令牌桶

令牌桶的算法特点：

好处：可以方便地应对突发出口流量。

比如：可以改变令牌发放速度(需要后端系统能力的提升)，算法能按照新的发送速率调大令牌的发放数量，使得出口突发流量能被处理。

令牌生成的速度固定，消费速度不固定。

代码简单实现：

package ratelimit;


import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Logger;


public class TokenBucketLimiter {


    //桶的容量
    private static int capacity=100;
    //令牌生成速度 rate/s
    private static final int rate=50;
    //令牌数量
    private volatile static AtomicInteger tokens=new AtomicInteger(0);

    /**
     * 开启一个线程，按固定频率放入令牌桶固定个数
     */
    public static void productTokens(){
        ScheduledExecutorService scheduledExecutorService= Executors.newScheduledThreadPool(1);
        scheduledExecutorService.scheduleAtFixedRate(()->{
            int allTokens = tokens.get()+rate;
            //设置当前的tokens数量
            tokens.set(allTokens);
        },1000,1000,TimeUnit.MILLISECONDS);

    }


    /**
     * true是被限流了
     *
     * @param needCount
     * @return
     */
    public static synchronized boolean limited(int needCount){
        if(tokens.get()<needCount){

            return true;
        }

        System.out.println("此时令牌桶中数量有: "+tokens.getAndDecrement());
        return false;
    }
    public static void main(String[] args) {
        //开启生产者任务
        productTokens();
        //定义一个原子类，
        AtomicInteger atomicInteger=new AtomicInteger(0);
        ExecutorService executorService=Executors.newFixedThreadPool(5);
        for(int i=0;i<10000;i++){
            executorService.submit(()->{
                try {
                    Thread.sleep(200);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                //当前线程的名称
                String taskName=Thread.currentThread().getName();
                boolean isLimit=limited(1);
                //true被限流了
                if(isLimit){
                    System.out.println(taskName+"被限流了，累计限流次数: "+atomicInteger.incrementAndGet());
                }else {
                    System.out.println(taskName+"请求被正常处理了");
                }
            });

        }



    }




}

漏桶算法

漏桶(Leak Bucket) 算法限流的基本原理：

水(对应请求) 从进水口到漏桶里，漏桶以一定的速度出水(请求放行)，当水流速度过大，桶内的总水量大于桶容量会直接溢出，请求拒绝。

大致的规则如下：

1）进水口(对应客户端请求) 以任意速率流入漏桶。

2）漏桶的容量是固定的，出水(放行)速率也是固定的。

3）漏桶容量是不变的，如果处理速度太慢，桶内水的容量就会超出桶的容量，则后面的水滴会标识请求拒绝。

流程：

水(请求)先进入桶中，漏桶按照一定的速率进行漏水，如果漏桶满了，那么水就会溢出(请求拒绝)，可以看出来漏桶算法能强行限制数据的传输速率。

代码实现

package ratelimit;

import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

public class LeakBucketLimiter {
    //漏桶的容量
    private static int capacity=10;
    //漏水的速度 rate/s
    private static final int leakRate=5;
    //桶中水量
    private volatile static AtomicInteger waterLeaf=new AtomicInteger(0);

    /**
     * 开启一个线程，按固定频率放入令牌桶固定个数
     */
    public static void leakWater(){
        ScheduledExecutorService scheduledExecutorService= Executors.newScheduledThreadPool(1);
        scheduledExecutorService.scheduleAtFixedRate(()->{
            //现在桶中的水
            int water = waterLeaf.get()-leakRate;
            //设置当前的水量
            waterLeaf.set(Math.max(0,water));
        },1,1, TimeUnit.SECONDS);

    }
    public static synchronized boolean limited(int waterCount){
        if(waterLeaf.get()+waterCount>capacity){
            //水满了拒绝
            return true;
        }
        waterLeaf.addAndGet(waterCount);
        System.out.println("此时漏桶水量有: "+waterLeaf);
        return false;
    }
    public static void main(String[] args) {

        //开始漏水
        leakWater();

        //开启请求
        ScheduledExecutorService executorService=Executors.newScheduledThreadPool(5);
        AtomicInteger atomicInteger=new AtomicInteger(0);
        for(int i=0;i<10000;i++){
            executorService.submit(()->{
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                String taskName=Thread.currentThread().getName();
                boolean limited = limited(1);
                if(limited){
                    System.out.println(taskName+"请求被拦截，累计拦截次数"+atomicInteger.incrementAndGet());
                }else{
                    System.out.println(taskName+"请求访问成功！！！");
                }
            });

        }



    }


}