做项目的时候突然想到的问题:限流组件的应用和设计原则

做业务的同学都知道，在现实情况中，往往会出现流量暴增的情况。这些流量可能来自于黑客的爬虫，也可能来自于节日大促，或者其他一些渠道。当然业界都有对策，比如反爬、熔断、降级、限流等等不一而足。

我们今天就来谈谈其中的限流。

先看看业界常用的限流组件：

单机版JDK

• 自带的锁、信号量等
• Guava的RateLimiter

分布式

• Redis：Redis-cell、Redisson（基于lua脚本实现）
• Sentinel
• Hystrix

这里我用单机和分布式的维度简单划分了一下。其中有一些你应该见过。

那么什么时候用单机限流，什么时候用分布式限流呢？

其实要回答这个问题，首先要明确你的业务要限流的对象是什么。比如你的服务是单体的，那其实用单机限流正合适。当然现在这个时代的单体业务不多了，只有一些小项目用得比较多。

还有一种情况就是不需要在业务层面精确限流，比如说我们的业务部署在多个机器或者容器上，对外以Http的形式暴露服务，并且在Nginx层做了一定的负载均衡，导致流量会比较均匀地分布到各台机器上，此时用单机限流也是不错的：只要每台机器的流量都限制住，整体的流量就是被限制的。

注意这里是近似，试想，如果我们的机器数量扩容了一倍，那整体的限流阈值就会增长一倍。所以说，单机限流阈值也常用来保护机器不被打崩，考虑的角度更多是机器的性能，而非细化到某个业务接口。

如果要精确限制某个业务接口的流量（服务暴露形式不限，可能是Http，也可能是RPC），在分布式部署的环境下就需要采用分布式限流的方案了。

国内少数公司采用Netflix开源的Hystrix实现限流功能，还有一些公司直接引入了阿里开源的Sentinel来用，还有一些公司会使用Redisson提供的限流能力，或者直接编写Lua脚本实现。当然，还有一些大厂会自研限流组件，更好地满足自身业务需求。

那么问题来了，如果让你设计一个限流组件，要考虑哪些因素呢？这里给出一些参考：

1、限流的维度是什么？这个问题直接决定了限流功能要在哪个层面来实现。比如针对Http接口，限流器可以配置在Nginx层，进行域名限流。作为面向用户端防火墙的一个基础组件。也可以下放到业务接口层，限制某个业务的流量。

再进一步，如果我们要根据特定的接口参数进行限流（比如限制每个用户在一段时间内请求某个接口的频率），那限流组件就需要在接口层面实现了。

2、限流用什么算法？常见的有固定窗口法、漏斗法，以及令牌桶法，还有一些组件实现了具有预热功能的算法。现实场景中，出于应对突发流量的考虑，令牌桶算法的应用更为广泛。这个问题网上谈得比较多，不再赘述了。

我们这里贴一段Redisson中利用Lua脚本实现令牌桶算法的例子，感受一下：

## Redisson源码片段[Java]：获取令牌。

private <T> RFuture<T> tryAcquireAsync(RedisCommand<T> command, Long value) {
        return commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, command,
                "local rate = redis.call('hget', KEYS[1], 'rate');"
              + "local interval = redis.call('hget', KEYS[1], 'interval');"
              + "local type = redis.call('hget', KEYS[1], 'type');"
              + "assert(rate ~= false and interval ~= false and type ~= false, 'RateLimiter is not initialized')"

              + "local valueName = KEYS[2];"
              + "local permitsName = KEYS[4];"
              + "if type == '1' then "
                  + "valueName = KEYS[3];"
                  + "permitsName = KEYS[5];"
              + "end;"

              + "assert(tonumber(rate) >= tonumber(ARGV[1]), 'Requested permits amount could not exceed defined rate'); "

              + "local currentValue = redis.call('get', valueName); "
              + "if currentValue ~= false then "
                     + "local expiredValues = redis.call('zrangebyscore', permitsName, 0, tonumber(ARGV[2]) - interval); "
                     + "local released = 0; "
                     + "for i, v in ipairs(expiredValues) do "
                          + "local random, permits = struct.unpack('fI', v);"
                          + "released = released + permits;"
                     + "end; "

                     + "if released > 0 then "
                          + "redis.call('zremrangebyscore', permitsName, 0, tonumber(ARGV[2]) - interval); "
                          + "currentValue = tonumber(currentValue) + released; "
                          + "redis.call('set', valueName, currentValue);"
                     + "end;"

                     + "if tonumber(currentValue) < tonumber(ARGV[1]) then "
                         + "local nearest = redis.call('zrangebyscore', permitsName, '(' .. (tonumber(ARGV[2]) - interval), '+inf', 'withscores', 'limit', 0, 1); "
                         + "return tonumber(nearest[2]) - (tonumber(ARGV[2]) - interval);"
                     + "else "
                         + "redis.call('zadd', permitsName, ARGV[2], struct.pack('fI', ARGV[3], ARGV[1])); "
                         + "redis.call('decrby', valueName, ARGV[1]); "
                         + "return nil; "
                     + "end; "
              + "else "
                     + "redis.call('set', valueName, rate); "
                     + "redis.call('zadd', permitsName, ARGV[2], struct.pack('fI', ARGV[3], ARGV[1])); "
                     + "redis.call('decrby', valueName, ARGV[1]); "
                     + "return nil; "
              + "end;",
                Arrays.asList(getRawName(), getValueName(), getClientValueName(), getPermitsName(), getClientPermitsName()),
                value, System.currentTimeMillis(), ThreadLocalRandom.current().nextLong());

3、限流数据要保存在哪里？这个问题的答案依赖于具体的实现方案。比如我们如果用基于Redis发展出来的组件（比如Redis-cell、Redisson）来实现，限流数据就是存在Redis服务器中的。而如果采用Sentinel实现，限流数据就是存在内存中的。

4、限流数据量的控制。这个问题的解决方案依赖于限流对象的数量。如果是针对有限的几个接口做限流，数据量小到几乎可以不用考虑。

但如果是前面提到的“根据特定的业务参数进行限流”这种场景，就可能出现问题：比如针对用户ID做限流，那可能需要保存对应量级的限流数据（每个正在访问的用户都要记录访问频次）。

如果设计不恰当的话，内存很快就会上涨甚至被打爆。不信的话我们可以估算一下，如果用Redis实现，结合了业务属性的Redis keys一般要占用几百字节左右，那么1千万个用户就需要占用几个GB的空间。如果换到保存在内存中，一条请求用的限流对象同样可能也要占用几百字节。如果我们放任这些数据无限增加的话，后果可能是灾难性的。

所以你一定想到了应对办法，那就是要对老数据做过期删除处理。具体到实现的话，Redis可以设置keys的过期时间，让老keys过期后自动删除。内存中可以设置最多保存多少条限流数据，超过阈值时触发老数据淘汰机制，最常用的是LRU算法。

当然，过期时间或者内存容量上限，都需要根据业务实际情况进行制定。PS：实际上Sentinel就利用了Google开源的ConcurrentLinkedHashMap，利用它实现了LRU：

## Sentinel源码片段[Java]：利用ConcurrentLinkedHashMap实现LRU，可以淘汰老数据。

public class ConcurrentLinkedHashMapWrapper<T, R> implements CacheMap<T, R> {

    private final ConcurrentLinkedHashMap<T, R> map;

    public ConcurrentLinkedHashMapWrapper(long size) {
        if (size <= 0) {
            throw new IllegalArgumentException("Cache max capacity should be positive: " + size);
        }
        this.map = new ConcurrentLinkedHashMap.Builder<T, R>()
            .concurrencyLevel(DEFAULT_CONCURRENCY_LEVEL)
            .maximumWeightedCapacity(size)
            .weigher(Weighers.singleton())
            .build();
    }
    ...
}

5、时钟回拨情况的处理。这个问题一般在精细限流的场景更容易出现，比如限制几个毫秒内只能通过几个请求。

一旦NTP服务器同步出现抖动，或者服务器本地时间被人为修改，就可能会导致获取令牌算法出现错误，进而导致限流算法失效。

常见的解决思路是让限流服务器重新获取一次时间，避免多个请求端的时间不一致。Redis-cell源码中就采用了这一方案：

## Redis-cell源码[rust]：获取令牌前，主动同步当前时间。    

    /// Gets the given key's value and the current time as dictated by the
    /// store (this is done so that rate limiters running on a variety of
    /// different nodes can operate with a consistent clock instead of using
    /// their own). If the key was unset, -1 is returned.
    fn get_with_time(&self, key: &str) -> Result<(i64, time::Tm), CellError>;

    fn get_with_time(&self, key: &str) -> Result<(i64, time::Tm), CellError> {
        match self.map.get(key) {
            Some(n) => Ok((*n, time::now_utc())),
            None => Ok((-1, time::now_utc())),
        }
    }

    // 下面的代码是rust库函数，获取时间

    /// Returns the current time in UTC
    pub fn now_utc() -> Tm {
        at_utc(get_time())
    }

    /**
     * Returns the current time as a `timespec` containing the seconds and
     * nanoseconds since 1970-01-01T00:00:00Z.
     */
    pub fn get_time() -> Timespec {
        let (sec, nsec) = sys::get_time();
        Timespec::new(sec, nsec)
    }

    pub fn get_time() -> (i64, i32) {
        let mut tv = libc::timespec { tv_sec: 0, tv_nsec: 0 };
        unsafe { libc::clock_gettime(libc::CLOCK_REALTIME, &mut tv); }
        (tv.tv_sec as i64, tv.tv_nsec as i32)
    }

从以上几点来看，要想设计一个好用且合格的限流组件，还真不是件容易的事情。

最后，本文也是实际工作中的一些经验之谈，很多细节尚没有讲到。有兴趣的同学可继续深究各种限流组件的实现原理，如有其他观点，欢迎交流。

来源：http://www.cnblogs.com/xiaoxi666