背景说明

限流在微服务架构的系统中是十分常见的需求，同样限流也是各个服务接入阶段也是不得不考虑的问题。在面对流量激增的情况下，我们要考虑给程序做限流控制，避免短时间过多的流量直接击垮系统。

所谓的限流指的是什么呢？可以这么理解，在一个时间周期内请求的次数，不得超过我们给定的一个预期值。如果超过了必须做出拒绝的策略，拒绝策略可以是丢弃，或者阻塞。那么这种需求的常见的实现方案有哪些呢？

常见的实现方案

限流功能常见的实现方案包括：

• 固定窗口法。
• 滑动窗口法
• 滑动日志法
• 漏桶法
• 令牌法

对于固定窗口和滑动窗口的理解

假定我们的限流要求是4秒内只能通过20个请求，那么固定窗口是将4秒的时间跨度视作一个整体窗口。要求在这一个窗口内，请求的次数不得超过20次。并且当时间经过4秒后，整个窗口会重置，窗口的计数器置为0。

这种做法在实现上很简单，但是会有一些问题无法很好的处理，比如窗口边界发生请求突增导致限流的结果超过预期设置的QPS。怎么理解这个现象呢？比如周期的最后一秒交易数量到了20次，在下一周期的第1秒交易数量也达到了20次。那么将两部分合起来，它的QPS其实是超过我们的预期。

为了解决边界请求突增的问题，我们可以使用滑动窗口法。滑动窗口法的一个核心思想是，它基于固定窗口法做了改进，固定窗口的问题是将整个时间跨度视为一个窗口，划分的粒度太粗了。那么滑动窗口呢，它的做法是划分出多个窗口，每个窗口内各自持有一个计数器。还是在刚刚的场景，划分出4个窗口，4个窗口各自都持有计数器。这种做法就解决了刚刚固定窗口的问题，在第4秒虽然来了20个请求，那么属于第4秒的窗口的计数器达到了20。现在时间来到4.5秒，这个时候第4秒的窗口是不会被回收的。那么4.5秒的时候进来的交易同样会被限流住，故此问题解决。滑动窗口就是通过这种方式让限流效果更加的平滑。

结合漏桶和滑动日志的特点实现限流器

漏桶中只能以恒定的速率剔除过期元素，而滑动日志的做法中让每一个元素都记录了请求的时间，那么我可以将这两者结合起来。在剔除的阶段，我可以拿到每个元素的请求时间并计算出过期时间，与当前时间进行实时比较，这样就能准确剔除真正到期了的元素。

通过这种方式，能够应对流量突增的情况。

下面附上我实现的限流算法的代码。

package com.hhdd.demos.rateLimitter;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalTime;
import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 简单的限流器实现
 * 实现思想和漏桶思路有点像，但略有变形
 * 漏桶是按照恒定速率清除桶内的元素
 * 本实现因桶内记录了时间戳故可更灵活的剔除过期成员
 *
 * @Author huanghedidi
 * @Date 2025/2/7 21:12
 */
public class QueueTimeLogRateLimiter<T extends RateLimiterRequest> {
    /**
     * 限流器的名称，如针对每个接口设置一个限流器
     * 则限流器名称可赋值为接口名称
     */
    private String name;
    /**
     * 漏桶思想中桶的容积
     */
    private int capacity;
    /**
     * 时间跨度，单位为ms
     */
    private long timeSpan;
    /**
     * 漏桶内的元素用queue记录
     * queue数据结构是有序的，方便剔除过期元素
     */
    private Queue<T> queue;
    /**
     * 限流器是共享资源，存在并发问题，操作需要用锁
     */
    private Lock lock;

    private AtomicInteger successCnt = new AtomicInteger(0);

    public QueueTimeLogRateLimiter(String name, int capacity, long timeSpan) {
        this.name = name;
        this.capacity = capacity;
        this.timeSpan = timeSpan;
        this.queue = new LinkedList<>();
        this.lock = new ReentrantLock();
    }

    public int getSuccessCnt() {
        return this.successCnt.get();
    }

    public void offer(T item) {
        try {
            lock.lock();
            //            print("开始处理！");
            cleanDueElement();
            if (this.capacity <= this.queue.size()) {
                throw new RuntimeException("服务器负载已经饱和，请稍后再试！！");
            }
            this.queue.offer(item);
            //            print("入队成功！");
            successCnt.incrementAndGet();
        } catch (Exception e) {
            //            e.printStackTrace();
            throw new RuntimeException(e.getMessage());
        } finally {
            lock.unlock();
        }

    }

    /**
     * 清除过期元素
     */
    public void cleanDueElement() {
        if (!queue.isEmpty()) {
            long now = System.currentTimeMillis();
            boolean res = queue.peek().getRequestTime() + this.timeSpan < now;
            String log = String.format("开始清理任务！queue内的元素个数为：%s\n，当前时间为" +
                                       "：%s\n，queue中第一个元素的请求时间为：%s\n，队内第一个元素可以出队吗: %s", this.queue.size(), now, queue.peek().getRequestTime(), res);
            //            print(log);
            while (!queue.isEmpty() && queue.peek().getRequestTime() + this.timeSpan < now) {
                queue.poll();
            }
        }
    }

    /**
     * 测试
     * @param args
     */
    public static void main(String[] args) {
        int threadNum = 5;
        long start = System.currentTimeMillis();
        QueueTimeLogRateLimiter<RateLimiterRequest> rateLimiter = new QueueTimeLogRateLimiter<>("hhdd", 2, 1000L);
        for (int i = 0; i < threadNum; i++) {
            new Thread(() -> {
                while (true) {
                    try {
                        RateLimiterRequest request = new RateLimiterRequest();
                        rateLimiter.offer(request);
                    } catch (Exception e) {
                        print(e.getMessage());
                    } finally {
                        // 输出统计信息
                        getStaticInfo(start, System.currentTimeMillis(), rateLimiter.getSuccessCnt());
                        try {
                            TimeUnit.MILLISECONDS.sleep(1);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                }
            }).start();
        }
    }

    /**
     * 获取统计数据
     */
    private static void getStaticInfo(long start, long end, int cnt) {
        BigDecimal timeSpan = new BigDecimal(end - start);
        BigDecimal requestCnt = new BigDecimal(cnt);
        BigDecimal qps = requestCnt.multiply(new BigDecimal(1000)).divide(timeSpan, 2, RoundingMode.HALF_UP);
//        double qps = (cnt) * 1000.0 / (end - start);
        String log = String.format("起止时间跨度为%s毫秒内，成功次数：%s，QPS：%s", end - start, cnt, qps);
        print(log);
    }

    public static void print(String msg) {
//        long now = System.currentTimeMillis();
        LocalTime now = LocalTime.now();
        String name = Thread.currentThread().getName();
        String log = String.format("%s---%s--:[msg]%s", now + "", name, msg);
        System.out.println(log);
    }

}

//-----------------------------------------------------------------------------
package com.hhdd.demos.rateLimitter;

/**
 * 用来封装限流的请求组件
 *
 * @Author huanghedidi
 * @Date 2025/2/7 21:10
 */
public class RateLimiterRequest {

    private long requestTime;

    public RateLimiterRequest() {
        this.requestTime = System.currentTimeMillis();
    }

    public long getRequestTime() {
        return requestTime;
    }
}

在上面代码中也写了测试案例，

在案例中我启动了5个线程并发对限流器发送请求，限流器配置的QPS为2

我们看下限流器的限流效果

上面我截取了程序刚开始运行阶段和经过七八秒后的结果进行展示，我们能够看到一开始的时候QPS比较高，但是随着程序运行QPS逐渐收敛到我们预期的结果。

针对这个现象我们分析其中的原因也很容易理解，因为我的算法的实现支持流量突增，那么在流量突增的短期的时间跨度内QPS会非常的高，超出我们设置的预期。但如果以1s这个时间范围为单位，QPS必会收敛到预期的2。并且随着程序的不断执行，突增的现象会越来越不明显。从程序的整体的运行周期来看，QPS会收敛到预期的目标！

局限性分析

我上面的实现方案也有其局限性。具体如下：

• 考虑一种极端的场景，假设我要求1ms内限流10笔，这种情况会退化成固定窗口模式，因为1ms内发生的交易其请求时间相同失效时间相同，视作同一窗口，则固定窗口模式下的边界问题也是会有的。此方式在极端情况下也是有边界问题，但概率大大较低，能以更平滑的方式处理限流。
• 漏桶算法有一个很大的特点是整流，能够让后续的流量严格按照漏出的速率为上限。而我的实现方案做不到这一点，因为剔除的速率是动态调整的，那么这种特性在有些时候可能也会带来一些问题。

参考文章

Java实现滑动窗口限流策略：全面解析与代码示例-腾讯云开发者社区-腾讯云

https://juejin.cn/post/7139191682138341383

面试官：来，年轻人！请手撸5种常见限流算法！ - Java填坑笔记 - 博客园

5种限流算法，7种限流方式，挡住突发流量？ | 未读代码