java限流算法

作者: 水煮鱼又失败了 | 来源:发表于2021-05-31 22:13 被阅读0次

1 场景

程序中经常需要对接口进行限流，防止访问量太大，导致程序崩溃。

常用的算法有：计数算法、漏桶算法、令牌桶算法，最常用的算法是后面两种。

2 算法详解

2.1 计数算法

2.1.1 说明

技术算法，为最简单的限流算法。

核心思想是，每隔一段时间，为计数器设定最大值，请求一次，计数器数量减一，如果计数器为0，则拒绝请求。

2.1.2 图示

计数器算法.png

2.1.3 适用场景

虽然此算法是大多数人第一个想到可以限流的算法，但是不推荐使用此算法。

因为，此算法有个致命性的问题，如果1秒允许的访问次数为100，前0.99秒内没有任何请求，在最后0.01秒内，出现了200个请求，则这200个请求，都会获取调用许可，给程序带来一次请求的高峰。

如下图所示：

计数器算法缺点.png

2.1.4 代码

import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;

/**
 * 计数器限流器
 */
public class CountLimiter {
    /**
     * 执行区间(毫秒)
     */
    private int secondMill;
    
    /**
     * 区间内计数多少次
     */
    private int maxCount;
    
    /**
     * 当前计数
     */
    private int currentCount;
    
    /**
     * 上次更新时间（毫秒）
     */
    private long lastUpdateTime;
    
    public CountLimiter(int second, int count) {
        if (second <= 0 || count <= 0) {
            throw new IllegalArgumentException("second and time must by positive");
        }
        this.secondMill = second * 1000;
        this.maxCount = count;
        this.currentCount = this.maxCount;
        this.lastUpdateTime = System.currentTimeMillis();
    }
    
    /**
     * 刷新计数器
     */
    private void refreshCount() {
        long now = System.currentTimeMillis();
        if ((now - this.lastUpdateTime) >= secondMill) {
            this.currentCount = maxCount;
            this.lastUpdateTime = now;
        }
    }
    
    /**
     * 获取授权
     * @return
     */
    public synchronized boolean tryAcquire() {
        // 刷新计数器
        this.refreshCount();
        if ((this.currentCount - 1) >= 0) {
            this.currentCount--;
            return true;
        } else {
            return false;
        }
    }
}

测试方法：

public static void main(String[] args) throws Exception {
    // 1秒限制执行2次
    CountLimiter countLimiter = new CountLimiter(1, 2);
    for (int i = 0; i < 10; i++) {
        System.out.println(LocalDateTime.now() + " " + countLimiter.tryAcquire());
        TimeUnit.MILLISECONDS.sleep(200);
    }
}

执行结果：

2021-05-31T22:01:08.660 true
2021-05-31T22:01:08.868 true
2021-05-31T22:01:09.074 false
2021-05-31T22:01:09.275 false
2021-05-31T22:01:09.485 false
2021-05-31T22:01:09.698 true
2021-05-31T22:01:09.901 true
2021-05-31T22:01:10.104 false
2021-05-31T22:01:10.316 false
2021-05-31T22:01:10.520 false

2.2 漏桶算法

2.2.1 说明

漏桶算法称为leaky bucket，可限制指定时间内的最大流量，如限制60秒内，最多允许100个请求。

其中接受请求的速度是不恒定的（水滴入桶），处理请求的速度是恒定的（水滴出桶）。

算法总体描述如下：

有个固定容量的桶B（指定时间区间X，允许的的最大流量B），如60秒内最多允许100个请求，则B为100，X为60。
有水滴流进来（有请求进来），桶里的水+1。
有水滴流出去（执行请求对应的业务），桶里的水-1（业务方法，真正开始执行=>这是保证漏桶匀速处理业务的根本），水滴流出去的速度是匀速的，流速为B/X（1毫秒100/60次，约1毫秒0.00167次，精度可根据实际情况自己控制）
水桶满了后（60秒内请求达到了100次），水滴无法进入水桶，请求被拒绝

2.2.2 图示

实际开发中，漏桶的使用方式可参考下图：

需注意，水滴滴落的时候，才开始执行业务代码，而不是水滴进桶的时候，去执行业务代码。

业务代码的执行方式，个人认为有如下两种：

同步执行

1、调用方请求时，如水滴可以放入桶中，调用方所在的线程“阻塞”
2、水滴漏出时，唤醒调用方线程，调用方线程，执行具体业务
异步执行

1、调用方请求时，如水滴可以放入桶中，调用方所在的线程收到响应，方法将异步执行
2、水滴漏出时，水桶代理执行具体业务

网上很多滴桶的实现代码，在水滴进桶的时候，就去执行业务代码了。这样会导致业务代码，无法匀速地执行，仍然对被调用的接口有一瞬间流量的冲击（和令牌桶算法的最终实现效果一样）。

漏桶算法.png

2.2.3 适用场景

水桶的进水速度是不可控的，有可能一瞬间有大量的请求进入水桶。处理请求的速度是恒定的（滴水的时候处理请求）。

此算法，主要应用于自己的服务，调用外部接口。以均匀的速度调用外部接口，防止对外部接口的压力过大，而影响外部系统的稳定性。如果影响了别人的系统，接口所在公司会来找你喝茶。

漏桶算法，主要用来保护别人的接口。

2.2.4 代码

本实例代码的实现，在水滴滴下，执行具体业务代码时，采用同步执行的方式。即唤醒调用方的线程，让"调用者"所属的线程去执行具体业务代码，去调用接口。

import java.net.SocketTimeoutException;
import java.time.LocalDateTime;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;

/**
 * 漏桶算法
 */
public class LeakyBucketLimiterUtil {
    
    /**
     * 漏桶流出速率(多少纳秒执行一次)
     */
    private long outflowRateNanos;
    
    /**
     * 漏桶容器
     */
    private volatile BlockingQueue<Drip> queue;
    
    /**
     * 滴水线程
     */
    private Thread outflowThread;
    
    /**
     * 水滴
     */
    private static class Drip {
        /**
         * 业务主键
         */
        private String busId;
        
        /**
         * 水滴对应的调用者线程
         */
        private Thread thread;
        
        public Drip(String busId, Thread thread) {
            this.thread = thread;
        }
        
        public String getBusId() {
            return this.busId;
        }
        
        public Thread getThread() {
            return this.thread;
        }
    }
    
    /**
     * @param second 秒
     * @param time   调用次数
     */
    public LeakyBucketLimiterUtil(int second, int time) {
        if (second <= 0 || time <= 0) {
            throw new IllegalArgumentException("second and time must by positive");
        }
        
        outflowRateNanos = TimeUnit.SECONDS.toNanos(second) / time;
        queue = new LinkedBlockingQueue<>(time);
        
        outflowThread = new Thread(() -> {
            while (true) {
                Drip drip = null;
                try {
                    // 阻塞，直到从桶里拿到水滴
                    drip = queue.take();
                } catch (Exception e) {
                    e.printStackTrace();
                }
                if (drip != null && drip.getThread() != null) {
                    // 唤醒阻塞的水滴里面的线程
                    LockSupport.unpark(drip.getThread());
                }
                // 休息一段时间，开始下一次滴水
                LockSupport.parkNanos(this, outflowRateNanos);
            }
        }, "漏水线程");
        outflowThread.start();
    }
    
    /**
     * 业务请求
     *
     * @return
     */
    public boolean acquire(String busId) {
        Thread thread = Thread.currentThread();
        Drip drip = new Drip(busId, thread);
        if (this.queue.offer(drip)) {
            LockSupport.park();
            return true;
        } else {
            return false;
        }
    }
}

测试代码如下：

public static void main(String[] args) throws Exception {
    // 1秒限制执行1次
    LeakyBucketLimiterUtil leakyBucketLimiter = new LeakyBucketLimiterUtil(5, 2);
    for (int i = 0; i < 10; i++) {
        new Thread(new Runnable() {
            @Override
            public void run() {
                String busId = "[业务ID：" + LocalDateTime.now().toString() + "]";
                if (leakyBucketLimiter.acquire(busId)) {
                    System.out.println(LocalDateTime.now() + " " + Thread.currentThread().getName() + "：调用外部接口...成功：" + busId);
                } else {
                    System.out.println(LocalDateTime.now() + " " + Thread.currentThread().getName() + "：调用外部接口...失败：" + busId);
                }
            }
        }, "测试线程-" + i).start();
        TimeUnit.MILLISECONDS.sleep(500);
    }
}

执行结果如下：

2021-05-31T20:52:52.297 测试线程-0：调用外部接口...成功：[业务ID：2021-05-31T20:52:52.295]
2021-05-31T20:52:53.782 测试线程-3：调用外部接口...失败：[业务ID：2021-05-31T20:52:53.782]
2021-05-31T20:52:54.286 测试线程-4：调用外部接口...失败：[业务ID：2021-05-31T20:52:54.286]
2021-05-31T20:52:54.799 测试线程-1：调用外部接口...成功：[业务ID：2021-05-31T20:52:52.761]
2021-05-31T20:52:55.300 测试线程-6：调用外部接口...失败：[业务ID：2021-05-31T20:52:55.300]
2021-05-31T20:52:55.806 测试线程-7：调用外部接口...失败：[业务ID：2021-05-31T20:52:55.806]
2021-05-31T20:52:56.307 测试线程-8：调用外部接口...失败：[业务ID：2021-05-31T20:52:56.307]
2021-05-31T20:52:56.822 测试线程-9：调用外部接口...失败：[业务ID：2021-05-31T20:52:56.822]
2021-05-31T20:52:57.304 测试线程-2：调用外部接口...成功：[业务ID：2021-05-31T20:52:53.271]
2021-05-31T20:52:59.817 测试线程-5：调用外部接口...成功：[业务ID：2021-05-31T20:52:54.799]

2.3 令牌桶算法

2.3.1 说明

令牌桶算法，主要是匀速地增加可用令牌，令牌数因为桶的限制有数量上限。

请求拿到令牌，相当于拿到授权，即可进行相应的业务操作。

2.3.2 图示

令牌桶算法.png

2.3.3 适用场景

和漏桶算法比，有可能导致短时间内的请求数上升（因为拿到令牌后，就可以访问接口，有可能一瞬间将所有令牌拿走），但是不会有计数算法那样高的峰值（因为令牌数量是匀速增加的）。

一般自己调用自己的接口，接口会有一定的伸缩性，令牌桶算法，主要用来保护自己的服务器接口。

2.3.4 代码

简易代码实现如下：

import java.time.LocalDateTime;
import java.util.concurrent.TimeUnit;

/**
 * 令牌桶限流算法
 */
public class TokenBucketLimiter {
    
    /**
     * 桶的大小
     */
    private double bucketSize;
    
    /**
     * 桶里的令牌数
     */
    private double tokenCount;
    
    /**
     * 令牌增加速度(每毫秒)
     */
    private double tokenAddRateMillSecond;
    
    /**
     * 上次更新时间（毫秒）
     */
    private long lastUpdateTime;
    
    /**
     * @param second 秒
     * @param time   调用次数
     */
    public TokenBucketLimiter(double second, double time) {
        if (second <= 0 || time <= 0) {
            throw new IllegalArgumentException("second and time must by positive");
        }
        // 桶的大小
        this.bucketSize = time;
        // 桶里的令牌数
        this.tokenCount = this.bucketSize;
        // 令牌增加速度(每毫秒)
        this.tokenAddRateMillSecond = time / second / 1000;
        // 上次更新时间（毫秒）
        this.lastUpdateTime = System.currentTimeMillis();
    }
    
    /**
     * 刷新桶内令牌数(令牌数不得超过桶的大小)
     * 计算“上次刷新时间”到“当前刷新时间”中间，增加的令牌数
     */
    private void refreshTokenCount() {
        long now = System.currentTimeMillis();
        this.tokenCount = Math.min(this.bucketSize, this.tokenCount + ((now - this.lastUpdateTime) * this.tokenAddRateMillSecond));
        this.lastUpdateTime = now;
    }
    
    /**
     * 尝试拿到权限
     *
     * @return
     */
    public synchronized boolean tryAcquire() {
        // 刷新桶内令牌数
        this.refreshTokenCount();
        if ((this.tokenCount - 1) >= 0) {
            // 如果桶中有令牌，令牌数-1
            this.tokenCount--;
            return true;
        } else {
            // 桶中已无令牌
            return false;
        }
    }
}

测试代码：

public static void main(String[] args) throws Exception{
    // 2秒执行1次
    TokenBucketLimiter leakyBucketLimiter = new TokenBucketLimiter(2, 1);
    for (int i = 0; i < 10; i++) {
        System.out.println(LocalDateTime.now() + " " + leakyBucketLimiter.tryAcquire());
        TimeUnit.SECONDS.sleep(1);
    }
}

执行结果如下：

2021-05-31T21:38:34.560 true
2021-05-31T21:38:35.582 false
2021-05-31T21:38:36.588 true
2021-05-31T21:38:37.596 false
2021-05-31T21:38:38.608 true
2021-05-31T21:38:39.610 false
2021-05-31T21:38:40.615 true
2021-05-31T21:38:41.627 false
2021-05-31T21:38:42.641 true
2021-05-31T21:38:43.649 false

2.3.5 第三方工具类

可以使用Guava中的RateLimiter来实现令牌桶的限流功能。

maven依赖如下：

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>30.1.1-jre</version>
</dependency>

直接获取令牌（true为获取到令牌，false为获取失败）：

RateLimiter rateLimiter = RateLimiter.create(2);
boolean acquireResule = rateLimiter.tryAcquire();
if (acquireResule) {
    System.out.println("获取令牌：成功");
} else {
    System.out.println("获取令牌：失败");
}

等待尝试获取令牌（阻塞当前线程，直到获取到令牌）：

RateLimiter rateLimiter = RateLimiter.create(2);
// 阻塞获取令牌
double waitCount = rateLimiter.acquire();
System.out.println("阻塞等待时间：" + waitCount);

java限流算法

1 场景

2 算法详解

2.1 计数算法

2.1.1 说明

2.1.2 图示

2.1.3 适用场景

2.1.4 代码

2.2 漏桶算法

2.2.1 说明

2.2.2 图示

2.2.3 适用场景

2.2.4 代码

2.3 令牌桶算法

2.3.1 说明

2.3.2 图示

2.3.3 适用场景

2.3.4 代码

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

java全栈

业务整理

架构

数据结构与算法

架构

算法

方案