分布式限流方案实现

分布式限流

一、需求背景：

1、纵横客业务调用口碑接口时（尤其是定时任务调接口提数），常常收到口碑网关返回isv.app-call-limited 超过调用频率限制，
目前仅仅是简单做了一下Thread.sleep(随机数)来实现平滑请求。
2、定时任务作为一个次级业务，当数据库压力过大时（尤其是业务高峰期），理论上应该减小定时任务的流量，为关键业务让路。
目前无法控制定时任务对数据库的压力
3、工作流启动后为多节点分布式数据计算，但是调用口碑为同一API接口，无法通过单机进行限流

二、需求目的

1、通过对调用口碑API接口的相关任务进行限流，来实现平滑调用，并且避免sleep引起的性能损耗
2、通过对数据库操作的代码块进行限流，来实现平滑调用，并且避免sleep引起的性能损耗
3、通过数据库压力的动态反馈，对数据库操作进行动态限流。动态实现错峰运行

三、实现方案

1、使用guava提供工具库里的RateLimiter类（单机限流）

内部采用令牌捅算法实现进行限流

2、redis+lua（分布式限流)

分布式限流最关键的是要将限流服务做成原子化。使用redis+lua实现某时间窗内某个接口的请求数限流

四、接口

     /**
     * 限流
     * @return true：限流阈值内；false：超过限流阈值
     * @see DistributionRatelimiter.DEFAULT_LIMIT
     */
    public boolean limit();
    /**
     * 限流
     * @param limit限流量
     * @see DistributionRatelimiter.DEFAULT_TIMEUNIT
     * @return true：限流阈值内；false：超过限流阈值
     */
    public boolean limit(int limit);
    /**
     * 限流
     * @param limit 限流量
     * @param unit 限流单位
     * @return true：限流阈值内；false：超过限流阈值
     */
    public boolean limit(int limit,TimeUnit unit);
        /**
     * 限流
     * @param method 指定方法限流
     * @param limit限流量
     * @param unit限流单位
     * @returntrue：限流阈值内；false：超过限流阈值
     */
    public boolean limit(String method,int limit,TimeUnit unit);

五、用法

配置

①引入配置文件

<import resource="classpath:bean/summonercloud-ratelimiter.xml" />

②依赖jar包

<dependency>
    <groupId>com.yunzong.summonercloud</groupId>
    <artifactId>summonercloud-ratelimiter</artifactId>
    <version>1.0.1.2374</version>
</dependency>

使用

③bean注入

@Resource
RedisRatelimiter dRatelimiter;

④方法调用

//例如：一分钟限流500
<!--核心代码片段-->
boolean result = dRatelimiter.limit(500, TimeUnit.MINUTES);
if(result) {
    //没有限流，处理业务。。。。
}else {
    //被限流，拒绝服务
}

六、思考

如何动态感知数据库压力并且动态采取限流量措施实现错峰运行？
   需要DBA沟通看看有什么方案。

七、附Lua脚本

--
--lua 下标从 1 开始
-- 限流 key
local key = KEYS[1]
-- 限流大小
local limit = tonumber(ARGV[1])

local MINUTES = "MINUTES"

-- 获取当前流量大小
local curentLimit = tonumber(redis.call('get', key) or "0")
local m = ARGV[2]
if curentLimit + 1 > limit then
    -- 达到限流大小 返回
    return 0;
else
    -- 没有达到阈值 value + 1
    redis.call("INCRBY", key, 1)
    if m == MINUTES then
        redis.call("EXPIRE", key, 120)
    else
        redis.call("EXPIRE", key, 2)
    end
    return curentLimit + 1
end