Distributed Rate Limter

Related: http://massivetechinterview.blogspot.com/2015/10/develop-api-rate-limit-throttling-client.html
http://afghl.github.io/2017/07/28/distributed-system-01-rate-limiting.html

通常限流的limit会设置为峰值的2倍以上。真出现有流量被限的话，也就总流量的不到10%（除非是非正常流量）。而且，限流通常是一个相对静态的设置，大部分情况下，限流起作用了说明我们的大部分请求都得到正确处理（否则，早就触发熔断了）。所以为了简单起见，选择服务器端限流吧。

分布式集群里，该怎样计数？

限流的粒度通常是集群级别的。比如限制一个集群的qps限流到1w qps，集群有10个节点部署。那么限流该怎样执行？

我原来以为一个集群会做一个redis的counter，让每个节点去incr这个counter，但发现这样做性能太差了：incr counter的操作是同步的。

正确的做法是：让每个节点知道自己的可以handle的qps，然后在节点内部各自实现counter的功能（可以在native内存中加计数器，在每个请求处理器先post process一下）。推送有几种方案：

• 让节点告诉配置中心自己的负载情况，在配置中心实时为每个节点分配权重。
• 给每个节点推送分配 1w / 10 = 1000 的qps限额。

显然第一个方案太难实现了。通常使用第二个方案，缺点很明显：平均主义。但workaround也很简单：让某个集群所有机器的配置都保持一致即可。

当集群的节点数改变时，限流大小是否需要动态变化？

当然需要，但这是另外一个话题：当一个节点被下线时，首先应该被服务治理中心知道。而限流大小的变化，是这个事件的其中一个回调。

被挡掉的流量作何处理？

既然限流发生在服务器端，是否可以让用户自定义一个fallback方法？否则抛出一个RateLimitException。调用方接收到这个异常又该如何处理？

http://blog.csdn.net/u013815546/article/details/63253495

• A类算法，是否决式限流。即如果系统设定限流方案是1分钟允许100次调用，那么真实请求1分钟调用200次的话，意味着超出的100次调用，得到的是空结果或者调用频繁异常。
• B类算法，是阻塞式限流。即如果系统设定限流方案是1分钟允许100次调用，那么真实请求1分钟调用200次的话，意味着超出的100次调用，会均匀安排到下一分钟返回。（当然B类算法，也可以立即返回失败，也可以达到否决式限流的效果）

B类算法，如Guava包提供的RateLimiter，内部其实就是一个阻塞队列，达到阻塞限流的效果。然后分布式场景下，有一些思路悄悄的发生了变化。多个模块之间不能保证相互阻塞，共享的变量也不在一片内存空间中。为了使用阻塞限流的算法，我们不得不将统计流量放到redis一类的共享内存中，如果操作是一系列复合的操作，我们还不能使用redis自带的CAS操作(CAS操作只能保证单个操作的原子性)或者使用中间件级别的队列来阻塞操作，显示加分布式锁的开销又是非常的巨大。最终选择放弃阻塞式限流，而在分布式场景下，仅仅使用redis+lua脚本的方式来达到分布式-否决式限流的效果。redis执行lua脚本是一个单线程的行为，所以不需要显示加锁，这可以说避免了加锁导致的线程切换开销。

RedisDistributeLock

https://www.cnblogs.com/AndyAo/p/8144049.html
https://my.oschina.net/giegie/blog/1525931

• Nginx接入层限流
  按照一定的规则如帐号、IP、系统调用逻辑等在Nginx层面做限流
• 业务应用系统限流
  通过业务代码控制流量这个流量可以被称为信号量，可以理解成是一种锁，它可以限制一项资源最多能同时被多少进程访问。

http://www.x64coding.com/archives/66
Guava RateLimiter

缺点：只能作用于JVM内部，集群规模变化时，无法做到整个集群的速率随之变化（其实也可以实现，不过需要通过集群速率进行单机限流速率计算，依赖实例的负载均衡，效果可能稍微没那么好）

public class RedisEasyRateLimiterImpl implements EasyRateLimiter {

 

    private static final Logger LOGGER = Logger.getLogger(RedisEasyRateLimiterImpl.class.getName());

 

    private String redisKey;

 

    private Integer qps;// 限流器速率，自行注入

 

    private RedisTemplate<String, Long> redisTemplate;// spring redis template，需要注意，由于需要与lua脚本交互，Key、Value需要使用String序列化

 

    private RedisScript<Long> redisScript;// spring redis script

 

    public void acquire(int permits) {

        Long currentTime = System.currentTimeMillis();

        //@TODO 调用TIME命令获取统一时间

        //redisTemplate.getConnectionFactory().getConnection().time();//如此获取会造成连接泄漏

        LOGGER.info("current time: " + currentTime);

        Long holdTime = redisTemplate.execute(redisScript, Collections.singletonList(redisKey),

                String.valueOf(currentTime), String.valueOf(qps), String.valueOf(permits), UUID.randomUUID().toString().replace("-", ""));

        LOGGER.info("holdTime ms: " + holdTime);

        if (holdTime > 0) {

            try {

                TimeUnit.MILLISECONDS.sleep(holdTime);// 当前线程睡眠等待

            } catch (InterruptedException e) {

                throw new RuntimeException(e);

            }

        }

    }

 

    // setter and getter...

}

    <bean id="redisTemplate"

          class="org.springframework.data.redis.core.RedisTemplate"

          p:connection-factory-ref="jedisConnectionFactory"

          p:keySerializer-ref="stringRedisSerializer"

          p:valueSerializer-ref="stringRedisSerializer"/>

 

    <bean id="easyRateLimiterScript" class="org.springframework.data.redis.core.script.DefaultRedisScript"

          p:location="classpath:META-INF/easyratelimiter.lua"

          p:resultType="java.lang.Long"/>

 

    <bean id="redisEasyRateLimiterImpl" class="com.luol.tools.impl.RedisEasyRateLimiterImpl">

        <property name="qps" value="2"/>

        <property name="redisKey" value="RedisEasyRateLimiter#1"/>

        <property name="redisScript" ref="easyRateLimiterScript"/>

        <property name="redisTemplate" ref="redisTemplate"/>

    </bean>

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20

local nextPermissionTime = redis.call('GET', KEYS[1]) local currentTime = tonumber(ARGV[1]) local qps = tonumber(ARGV[2]) local permits = tonumber(ARGV[3]) local blockingTime = 0   if nextPermissionTime then     nextPermissionTime = tonumber(nextPermissionTime)     if nextPermissionTime > currentTime then         blockingTime = nextPermissionTime - currentTime     else         nextPermissionTime = currentTime     end else     nextPermissionTime = currentTime end   nextPermissionTime = nextPermissionTime + 1 / qps * permits * 1000 redis.call('SET', KEYS[1], nextPermissionTime) return blockingTime