首页 > 数据库> > 基于Redis实现分布式限流

基于Redis实现分布式限流

2021-01-11 02:03:47 作者：互联网

前言
模板方法
- 代码
方法一固定时间段限流
方法二时间段不固定限流
方法四利用有序集合实现滑动窗口
方法三令牌桶算法在Redis中的实现

前言

限流的作用大家其实都知道, 单机版本下的限流有Guava的RateLimiter以及JDK自带的Semaphore, 都能满足需求. 但是线上业务出于各种原因考虑, 还需要分布式的限流, 比如扩容/缩容时无法控制整个服务的控制速率等场景.

开源的分布式限流组件有阿里巴巴的Sentinel, 但其实我们还可以利用Redis来实现一种更加轻量级的限流

模板方法

参考AQS的独占锁设计, 定义一个基类, 规定好模板, 由子类实现核心逻辑tryAcquire, 父类根据tryAcquire的返回结果来实现不同逻辑

代码

public abstract class RateLimiter {

    private static final Logger LOGGER = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

    /**
     * redis key中附带的唯一标记
     */
    protected String name;

    protected String prefix;

    /**
     * 支持修改速率
     * 加上volatile是为了修改速率之后, 所有的线程都能感知到结果修改
     * 分布式环境下这个变量可以存储在分布式的配置中心里面
     */
    protected volatile int permitsPerSecond;

    public void setRate(int permitsPerSecond) {
        this.permitsPerSecond = permitsPerSecond;
    }

    public String getName() {
        return name;
    }

    public boolean acquire() throws InterruptedException {
        return acquire(1);
    }

    /**
     * 当获取不到permit(许可)时, 陷入睡眠, 睡眠时间由子类tryAcquire定义
     * 返回0时表示获取到了permit
     */
    public boolean acquire(int permits) throws InterruptedException {
        while (true) {
            long sleepTime = tryAcquire(permits);
            LOGGER.info("{} sleep {} ms", Thread.currentThread().getName(), sleepTime);
            if (sleepTime == 0) {
                return true;
            }
            // 异常抛出, 响应中断
            Thread.sleep(sleepTime);
        }
    }

    public abstract long tryAcquire();

    /**
     * 由子类实现, 方法逻辑为尝试获取许可, 获取不到则返回需要睡眠的时间
     */
    public abstract long tryAcquire(int permits);

}

方法一固定时间段限流

该方法是Redis官方推荐的一种方法, 具体逻辑为在Redis每秒生成一个key, key就是前缀+描级时间戳, value存储计数器, 当计数器中的数减到小于0时, 代表超出了并发限制. 且为每一个key设置10s的过期时间, 到期自动清除

优点

实现简单, 逻辑清晰
在对齐的窗口(时间片)上满足频率要求, 不对齐的时间片上最多是频率限制的2倍, 如果业务系统能满足这种波动, 那么问题不大

缺点

只能在固定时间片窗口内限流(1s), 无法做到滑动窗口限流
不适用于那种严格限制速率的场景
无法像Guava的RateLimiter那样, 支持预消费

核心逻辑

细节见代码与注释

public class FixedWindowRedisRateLimiter extends RateLimiter {

    public FixedWindowRedisRateLimiter(int permitsPerSecond) {
        this(UUID.randomUUID().toString(), permitsPerSecond);
    }

    public FixedWindowRedisRateLimiter(String name, int permitsPerSecond) {
        this.permitsPerSecond = permitsPerSecond;
        this.name = name;
        this.prefix = "F:RATE_LIMIT:";
    }

    private String getRedisKey(long currentSeconds) {
        return String.format("%s:%s:%s", prefix, name, currentSeconds);
    }

    @Override
    public long tryAcquire() {
        return tryAcquire(1);
    }

    @Override
    public long tryAcquire(int permits) {
        long epochSecond = Instant.now().getEpochSecond();
        // 睡眠到下一个时间窗口
        long sleepTime = 1000 * (epochSecond + 1) - System.currentTimeMillis();
        String redisKey = getRedisKey(epochSecond);
        // 核心逻辑
        RedisUtils.setNx(redisKey, String.valueOf(permitsPerSecond), 1000);
        long val = RedisUtils.decr(redisKey);
        // val>=0时, 表示获取到了permit
        if (val >= 0) return 0;
        return sleepTime;
    }

}

tryAcquire测试用例

使用CountDownLatch与CyclicBarrier配合完成并发测试

    @Test
    public void testTryAcquire1() throws Exception {
        int currentSize = 20;
        AtomicInteger atomicInteger = new AtomicInteger(currentSize);
        // 用于控制main线程结束
        CountDownLatch countDownLatch = new CountDownLatch(currentSize);
        // 用于控制并发, 一起发起请求
        CyclicBarrier cyclicBarrier = new CyclicBarrier(currentSize);

        RateLimiter rateLimiter = new FixedWindowRedisRateLimiter("custom", 3);
        for (int i = 0; i < currentSize; i++) {
            EXECUTORS.submit(new Runnable() {
                @Override
                public void run() {
                    for (int i = 0; i < 10; i++) {
                        try {
                            cyclicBarrier.await();
                        } catch (Exception e) {
                            // do nothing
                        }
                        LOGGER.info("{} tryAcquire result {}", Thread.currentThread().getName(), rateLimiter.tryAcquire());
                        int mark = atomicInteger.incrementAndGet();
                        // 分割线
                        if (mark % 20 == 0) LOGGER.info("********************************");
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            // do nothing
                        }
                    }
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await();
    }

tryAcquire测试用例结果

从结果来看符合预期, 在秒级时间戳窗口中, 一次只有三个线程获取到了permit

2021-01-10 21:31:46.693 pool-1-thread-14 tryAcquire result 920
2021-01-10 21:31:46.695 pool-1-thread-17 tryAcquire result 920
2021-01-10 21:31:46.692 pool-1-thread-6 tryAcquire result 920
2021-01-10 21:31:46.693 pool-1-thread-4 tryAcquire result 919
2021-01-10 21:31:46.693 pool-1-thread-5 tryAcquire result 920
2021-01-10 21:31:46.693 pool-1-thread-10 tryAcquire result 0
2021-01-10 21:31:46.693 pool-1-thread-8 tryAcquire result 0
2021-01-10 21:31:46.693 pool-1-thread-19 tryAcquire result 919
2021-01-10 21:31:46.692 pool-1-thread-11 tryAcquire result 920
2021-01-10 21:31:46.693 pool-1-thread-20 tryAcquire result 920
2021-01-10 21:31:46.693 pool-1-thread-12 tryAcquire result 919
2021-01-10 21:31:46.693 pool-1-thread-9 tryAcquire result 919
2021-01-10 21:31:46.693 pool-1-thread-16 tryAcquire result 920
2021-01-10 21:31:46.693 pool-1-thread-15 tryAcquire result 919
2021-01-10 21:31:46.693 pool-1-thread-3 tryAcquire result 920
2021-01-10 21:31:46.693 pool-1-thread-1 tryAcquire result 920
2021-01-10 21:31:46.693 pool-1-thread-7 tryAcquire result 920
2021-01-10 21:31:46.693 pool-1-thread-18 tryAcquire result 919
2021-01-10 21:31:46.693 pool-1-thread-2 tryAcquire result 0
2021-01-10 21:31:46.693 pool-1-thread-13 tryAcquire result 919
2021-01-10 21:31:46.699 ********************************
2021-01-10 21:31:47.709 pool-1-thread-2 tryAcquire result 0
2021-01-10 21:31:47.709 pool-1-thread-19 tryAcquire result 0
2021-01-10 21:31:47.709 pool-1-thread-8 tryAcquire result 0
2021-01-10 21:31:47.709 pool-1-thread-10 tryAcquire result 297
2021-01-10 21:31:47.709 pool-1-thread-12 tryAcquire result 297
2021-01-10 21:31:47.709 pool-1-thread-11 tryAcquire result 297
2021-01-10 21:31:47.709 pool-1-thread-6 tryAcquire result 296
2021-01-10 21:31:47.710 pool-1-thread-20 tryAcquire result 297
2021-01-10 21:31:47.710 pool-1-thread-4 tryAcquire result 296
2021-01-10 21:31:47.710 pool-1-thread-17 tryAcquire result 296
2021-01-10 21:31:47.710 pool-1-thread-9 tryAcquire result 296
2021-01-10 21:31:47.710 pool-1-thread-15 tryAcquire result 296
2021-01-10 21:31:47.710 pool-1-thread-5 tryAcquire result 296
2021-01-10 21:31:47.710 pool-1-thread-14 tryAcquire result 296
2021-01-10 21:31:47.710 pool-1-thread-13 tryAcquire result 296
2021-01-10 21:31:47.710 pool-1-thread-1 tryAcquire result 296
2021-01-10 21:31:47.710 pool-1-thread-18 tryAcquire result 296
2021-01-10 21:31:47.711 pool-1-thread-3 tryAcquire result 296
2021-01-10 21:31:47.711 pool-1-thread-16 tryAcquire result 296
2021-01-10 21:31:47.711 pool-1-thread-7 tryAcquire result 296
2021-01-10 21:31:47.711 ********************************

方法二时间段不固定限流

该方法也是Redis官网推荐的一种方法, 利用Redis的list数据结构, 也是在固定的时间段内进行限流, 有别于第一种方法的是选取的时间片不再是根据秒级时间戳, 具体见下面Redis官网的一段伪代码

Redis官网伪代码

这段伪代码存在并发问题, 所以实际使用中需要利用Lua脚本包装一下

FUNCTION LIMIT_API_CALL(ip)
current = LLEN(ip)
IF current > 10 THEN
    ERROR "too many requests per second"
ELSE
    IF EXISTS(ip) == FALSE
        MULTI
            RPUSH(ip,ip)
            EXPIRE(ip,1)
        EXEC
    ELSE
        RPUSHX(ip,ip)
    END
    PERFORM_API_CALL()
END

优点

相较于第一种方法, 该方法的优点在于list中可以存储ip等数据, 在特定场景下可能会有用. 以及优化了第一种方法中的按照秒级时间戳来选取时间段的方式, 改为不固定

缺点

第一种有的缺点, 第二种照样有
当QPS特别大时, list会膨胀, 按照一个value 1Byte来算, 30wQPS需要存储30w个value, 30w/1024, 接近300KB

核心逻辑

细节见代码与注释

public class SlidingWindowRedisRateLimiter extends RateLimiter {

    private static final String NEW_LINE = "\n";

    private static final String LUA_SCRIPT;

    // 越短越好
    private static final String PUSH_DATA = "VALUE";

    private static final long SUCCESS_VAL = -1;

    // 容错时间
    private static final long REDIS_NET_TIME = 3;

    // 核心逻辑就是这段lua脚本, 也就是上文提到的官网伪代码的lua代码版本
    static {
        StringBuilder sb = new StringBuilder();
        // -1表示成功, 因为无法区分llen为0是不存在了, 还是已经到0s了
        String returnSuccess = "return " + SUCCESS_VAL;

        // 数据定义
        sb.append("local key = KEYS[1]").append(NEW_LINE);
        sb.append("local limit = ARGV[1]").append(NEW_LINE);
        sb.append("local expireTime = ARGV[2]").append(NEW_LINE);
        sb.append("local pushData = ARGV[3]").append(NEW_LINE);
        sb.append("local permit = ARGV[4]").append(NEW_LINE);

        // len+permit大于limit
        sb.append("local len = redis.call('llen', key)").append(NEW_LINE);
        sb.append("if tonumber(len) + tonumber(permit) > tonumber(limit) then").append(NEW_LINE);
        sb.append("return redis.call('pttl', key)").append(NEW_LINE);
        sb.append("end").append(NEW_LINE);

        // 当key不存在时, 放入permit数量的数据, 然后设置过期时间
        sb.append("if tonumber(len) == 0 then").append(NEW_LINE);
        sb.append("for i = 0, permit - 1, 1 do").append(NEW_LINE);
        sb.append("redis.call('rpush', key, pushData)").append(NEW_LINE);
        sb.append("end").append(NEW_LINE);

        sb.append("redis.call('expire', key, expireTime)").append(NEW_LINE);
        sb.append(returnSuccess).append(NEW_LINE);
        sb.append("end").append(NEW_LINE);

        // key存在, 放入permit数量的数据
        sb.append("for i = 0, permit - 1, 1 do").append(NEW_LINE);
        sb.append("redis.call('rpush', key, pushData)").append(NEW_LINE);
        sb.append("end").append(NEW_LINE);
        sb.append(returnSuccess).append(NEW_LINE);
        LUA_SCRIPT = sb.toString();
    }

    public SlidingWindowRedisRateLimiter(int permitsPerSecond) {
        this(UUID.randomUUID().toString(), permitsPerSecond);
    }

    public SlidingWindowRedisRateLimiter(String name, int permitsPerSecond) {
        this.permitsPerSecond = permitsPerSecond;
        this.name = name;
        this.prefix = "S:RATE_LIMIT:";
    }

    private String getRedisKey() {
        return String.format("%s:%s", prefix, name);
    }

    @Override
    public long tryAcquire() {
        return tryAcquire(1);
    }

    @Override
    public long tryAcquire(int permits) {
        String[] keys = new String[]{getRedisKey()};
        String[] args = new String[]{String.valueOf(permitsPerSecond), String.valueOf(1), PUSH_DATA, String.valueOf(permits)};
        Long result = RedisUtils.eval(LUA_SCRIPT, keys, args, ScriptOutputType.INTEGER);
        if (Objects.equals(result, SUCCESS_VAL)) return 0;
        // 减去Redis可能得网络开销时间
        return result - REDIS_NET_TIME;
    }

}

tryAcquire测试用例

与方法一使用到的test case差不对, 都是使用CountDownLatch与CyclicBarrier配合完成并发测试

    @Test
    public void testAcquire2() throws InterruptedException {
        int currentSize = 20;
        AtomicInteger atomicInteger = new AtomicInteger(currentSize);
        CountDownLatch countDownLatch = new CountDownLatch(currentSize);
        CyclicBarrier cyclicBarrier = new CyclicBarrier(currentSize);

        RateLimiter rateLimiter = new SlidingWindowRedisRateLimiter("custom", 3);
        for (int i = 0; i < currentSize; i++) {
            EXECUTORS.submit(new Runnable() {
                @Override
                public void run() {
                    for (int i = 0; i < 10; i++) {
                        try {
                            cyclicBarrier.await();
                        } catch (Exception e) {
                            // do nothing
                        }
                        LOGGER.info("{} tryAcquire result {}", Thread.currentThread().getName(), rateLimiter.tryAcquire());
                        int mark = atomicInteger.incrementAndGet();
                        // 分割线
                        if (mark % 20 == 0) LOGGER.info("********************************");
                        try {
                            Thread.sleep(1000);
                        } catch (InterruptedException e) {
                            // do nothing
                        }
                    }
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await();
    }

tryAcquire测试用例结果

从结果来看符合预期, 一次只有三个线程获取到了permit

2021-01-11 01:14:30.783 pool-1-thread-4 tryAcquire result 0
2021-01-11 01:14:30.783 pool-1-thread-2 tryAcquire result 990
2021-01-11 01:14:30.783 pool-1-thread-6 tryAcquire result 991
2021-01-11 01:14:30.783 pool-1-thread-1 tryAcquire result 0
2021-01-11 01:14:30.783 pool-1-thread-17 tryAcquire result 992
2021-01-11 01:14:30.783 pool-1-thread-14 tryAcquire result 991
2021-01-11 01:14:30.783 pool-1-thread-11 tryAcquire result 0
2021-01-11 01:14:30.783 pool-1-thread-12 tryAcquire result 992
2021-01-11 01:14:30.783 pool-1-thread-16 tryAcquire result 994
2021-01-11 01:14:30.783 pool-1-thread-18 tryAcquire result 992
2021-01-11 01:14:30.783 pool-1-thread-9 tryAcquire result 990
2021-01-11 01:14:30.783 pool-1-thread-10 tryAcquire result 993
2021-01-11 01:14:30.783 pool-1-thread-8 tryAcquire result 994
2021-01-11 01:14:30.783 pool-1-thread-3 tryAcquire result 990
2021-01-11 01:14:30.783 pool-1-thread-20 tryAcquire result 992
2021-01-11 01:14:30.783 pool-1-thread-15 tryAcquire result 993
2021-01-11 01:14:30.783 pool-1-thread-7 tryAcquire result 993
2021-01-11 01:14:30.783 pool-1-thread-19 tryAcquire result 993
2021-01-11 01:14:30.783 pool-1-thread-5 tryAcquire result 991
2021-01-11 01:14:30.783 pool-1-thread-13 tryAcquire result 994
2021-01-11 01:14:30.785 ********************************
2021-01-11 01:14:31.794 pool-1-thread-19 tryAcquire result 0
2021-01-11 01:14:31.794 pool-1-thread-7 tryAcquire result 997
2021-01-11 01:14:31.794 pool-1-thread-13 tryAcquire result 996
2021-01-11 01:14:31.794 pool-1-thread-20 tryAcquire result 996
2021-01-11 01:14:31.794 pool-1-thread-3 tryAcquire result 995
2021-01-11 01:14:31.794 pool-1-thread-8 tryAcquire result 995
2021-01-11 01:14:31.794 pool-1-thread-9 tryAcquire result 995
2021-01-11 01:14:31.794 pool-1-thread-16 tryAcquire result 994
2021-01-11 01:14:31.794 pool-1-thread-6 tryAcquire result 0
2021-01-11 01:14:31.794 pool-1-thread-11 tryAcquire result 993
2021-01-11 01:14:31.794 pool-1-thread-5 tryAcquire result 0
2021-01-11 01:14:31.794 pool-1-thread-17 tryAcquire result 993
2021-01-11 01:14:31.794 pool-1-thread-12 tryAcquire result 994
2021-01-11 01:14:31.794 pool-1-thread-18 tryAcquire result 994
2021-01-11 01:14:31.794 pool-1-thread-10 tryAcquire result 995
2021-01-11 01:14:31.794 pool-1-thread-15 tryAcquire result 996
2021-01-11 01:14:31.795 pool-1-thread-14 tryAcquire result 993
2021-01-11 01:14:31.795 pool-1-thread-2 tryAcquire result 993
2021-01-11 01:14:31.795 pool-1-thread-1 tryAcquire result 993
2021-01-11 01:14:31.795 pool-1-thread-4 tryAcquire result 992
2021-01-11 01:14:31.795 ********************************

方法四利用有序集合实现滑动窗口

待更新

方法三令牌桶算法在Redis中的实现

待更新

标签：01,thread,Redis,tryAcquire,限流,result,2021,pool,分布式
来源： https://www.cnblogs.com/plumswine/p/14260365.html

基于Redis实现分布式限流

前言

模板方法

代码

方法一 固定时间段限流

优点

缺点

核心逻辑

tryAcquire测试用例

tryAcquire测试用例结果

方法二 时间段不固定限流

Redis官网伪代码

优点

缺点

核心逻辑

tryAcquire测试用例

tryAcquire测试用例结果

方法四 利用有序集合实现滑动窗口

方法三 令牌桶算法在Redis中的实现

方法一固定时间段限流

方法二时间段不固定限流

方法四利用有序集合实现滑动窗口

方法三令牌桶算法在Redis中的实现