如何应用 redis 实现限流

首发于 Dale’s blog

背景

在工作中时常会遇到须要对接口或者某个调用进行限流的状况。也会遇到在限流的同时对 redis 数据进行一些解决，在波及到分布式的情景下，就须要操作的原子性。

限流算法

支流的限流算法为以下四种：

1. 计数器（固定窗口）
2. 滑动窗口（宰割计数器）
3. 漏桶算法
4. 令牌桶算法

对于算法的解释，网上有很多好文章，在这里贴上罕用四种限流算法。

在本文中，探讨前两种也就是 计数器 以及 滑动窗口。

业务解释

限流，是在业务中常常遇到的场景。例如：对接口的限流、对调用的限流,etc。

以对接口限流为例，流程如下：

申请打到服务器之后，须要先判断以后接口是否达到阈值，若：

1. 达到阈值，则完结本次申请。
2. 未达到阈值，则 计数 ++，持续下一步调用。

限流能够有很多中方法，如果只是小型的单机部署利用，则能够思考在内存中进行计数与操作。若是简单的我的项目且分布式部署的我的项目，能够思考应用 redis 进行计数。且限流的逻辑不肯定要限于 Java 代码中，也能够应用 lua 在 nginx 进行操作，例如赫赫有名的 openresty，同理其余网关服务也可实现。

分布式业务中的限流

首先剖析业务场景，在分布式部署的 api 场景中须要留神以下几点：

1. 应用网关对 api 进行负载平衡，部署在不同服务器上的进行之间内存很难做到共享。
2. 基于限流的业务，是对整个零碎的某一个或者某一些接口进行限流，所以计数必须做到不同的过程都能够读取。
3. 对于计数的触发，是申请达到服务器上之后产生的，所以须要思考原子性。即：同一时刻，只有一个申请能够触发计数。这就对计数服务的要求提出了很高的并发要求。

剖析 nginx + lua 的可行性

nginx 罕用于申请的入口，在应用它的负载平衡之后，能够实现将申请散发到不同的服务上。应用 lua 对内存进行操作，仿佛能够实现上述要求（可行性待验证）。

然而，在理论状况中，一个零碎并肯定只会部署一个 nginx 作为入口。一方面是单机危险，另一方面是地理位置的不同，网络的不同对同一台机器的访问速度可能会有天差地别。所以，大家更喜爱应用 DNS 或者其余将申请达到多态 nginx 先做一层负载平衡。所以，单是 nginx + lua 并不能达到咱们的需要。

剖析 redis 的可行性

redis 是基于内存的一种非关系型数据库，它的并发是经得住考验的，同时它也能够满足不同过程对雷同数据读取、批改的需要。

对于原子性，redis 操作天生反对原子性，而且 string 类型的 INCR（原子累加）操作与 限流 业务又非常的符合。

redis 实现限流

让咱们再回到一开始的流程，计数限流的操作有：

1. 查问以后计数
2. 累加以后计数

在分布式系统中，必须要时刻留神 原子性。在繁多过程中，咱们保持数据线程平安的方法是加锁，无论是可重入锁还是 synchronized，其语义都是通知其余线程，这个数据（代码块）我当初征用了，你们等会再来。那在分布式系统中，咱们自然而然的能够想到 分布式锁。

伪代码如下：

Lock lock = getDistributedLock();

try{lock.lock();
    // 从 redis 中获取计数
    Integer count = getCountFromRedis();
    
    if(count >= limit){
        // 超过阈值，不予调用
        return false;
    }
    // 未超过阈值，容许调用
    incrRedisCount();
    return true;
}catch{...}finally{lock.unlock();
}

乍一看，这种逻辑没有问题，但其实问题很大：

1. 应用分布式锁显著会拖慢整个零碎，节约很多资源。
2. redis incr 操作会返回累加之后的值，所以查问操作是不必须的。

伪代码如下：

Integer count = incrRedisCount();
if (count >= limit){return false;}
return true;

是不是变的简略了很多。然而随之而来的有其余的问题，大部分的业务都不是要求咱们只对次数进行限度，更多的是要求咱们限度接口在一段时间内的申请次数 —- 滑动窗口。

滑动窗口的实现

顾名思义，滑动窗口就是将一个固定的窗口滑动起来。用于限流上来说就是，一段时间内进行计数，工夫一过，立马开始新的计数。
如何实现 一段时间 这个逻辑？
其实很简略，咱们齐全能够应用 工夫戳 来实现这一性能。

// 秒级工夫戳
long timestamp = System.currentTimeMillis() / 1000;
Long aLong = redisTemplate.opsForValue().increment(RedisKeyEnum.SYSTEM_FLOW_LIMIT.getKey() + timestamp);
return aLong;

此时会有一个问题，如果按以上代码来看，每秒创立一个键，那 redis 内存迟早会被撑爆。咱们须要一个策略来删除这个键。
笨的办法，能够记录这些键，而后异步去删除这些键。然而更好的办法是，在键第一次创立的时候设置一个稍大于窗口的过期值。所以，代码如下：

    /**
     * 按秒统计发送音讯数量
     *
     * @return
     */
    public Long getSystemMessageCountAtomic() {
        // 秒级工夫戳
        long timestamp = System.currentTimeMillis() / 1000;
        Long aLong = redisTemplate.opsForValue().increment(RedisKeyEnum.SYSTEM_FLOW_LIMIT.getKey() + timestamp);
        if (aLong != null && aLong == 1) {redisTemplate.expire(RedisKeyEnum.SYSTEM_FLOW_LIMIT.getKey() + timestamp, 2, TimeUnit.SECONDS);
        }
        return aLong;
    }

只有在第一次计数的时候才会执行 expire 命令。为什么须要设置稍大于窗口的工夫呢？
设想一下，如果设置和窗口一样的工夫，在 a 时刻的时候生成的键 keyA，而后过期工夫是一秒。而后在 b 时刻，生成的键也是 keyA（在同一秒内），然而因为网络或者其余起因，b 时刻的命令在一秒之后才发送到 redis server。因为过期工夫是一秒，此时旧的 keyA 曾经过期，那么 b 时刻就会创立一个新的键。

此时，须要思考另外一个问题，如果超过限度，以上代码会如何体现。

假如，一秒钟内只容许申请 100 次。那么第 101 次，也会去 redis 中执行 incr 命令，往后的申请都会执行。其实这些命令的执行时没有意义的，因为第 101 次时，这一秒的申请曾经到限度了，所以咱们须要另外一个存储来记录以上数据。

我选用 AtomicLong 来记录曾经到限的窗口。剖析一下是否可行。

1. AtomicLong 属于 java.util.concurrent.atomic 包，采纳 CAS 与 volatile 来保证数据的线程平安。
2. 上述需要，咱们只须要在单机上记录 flag 即可，不须要思考分布式状况。

阐述可行，以下展现代码。

private final AtomicLong flag = new AtomicLong();

/**
     * 零碎全局流量限度
     */
    public void systemFlowLimit() {
        // 判断 flag 是否与以后秒雷同
        if (flag.get() != System.currentTimeMillis() / 1000) {
            // 因为 flag.get 到 flag.set 之间的所有操作组合之后 不具备原子性，所以会有 小于 线程数 的线程会进入到这外面。// 意思是，当 第一个 线程将 flag 设置为 以后秒级 工夫戳之后，会有一部分线程曾经执行完 flag.get 的判断逻辑
            // 此时，局部线程会持续 redis 操作与 日志操作
            Long count = systemLimitService.getSystemMessageCountAtomic();
            if (count >= systemProperties.getFlowLimit()) {
                // 超过之后会将 flag 设置为以后秒
                flag.set(System.currentTimeMillis() / 1000);
                LOGGER.warn("system flow now is out of system flow limit,at:{}", System.currentTimeMillis() / 1000);
                throw new BusinessException(...);
            }
        } else {throw new BusinessException(...);
        }
    }

总结

以上整顿了应用 redis 做限流的一些办法，常常应用的算法便是滑动窗口，所以花了较大笔墨解释滑动窗口的实现。

当然，咱们还能够应用 lua 脚本来操作 redis 以实现限流与其余 redis 操作的配合。

我常常遇到的一个场景就是，往 redis 队列中写数据须要进行限流，当流量达到之后须要删除局部 redis 队列中的内容。此时，应用 lua 脚本来做能够很优雅的放弃多个 redis 操作的原子性，也能够缩小网络状况的开销。