龙空技术网

可以了,基于Redis和Lua实现分布式令牌桶限流

代码小当家 1829

前言:

今天各位老铁们对“nginx打印lua返回值”大体比较注重,姐妹们都需要分析一些“nginx打印lua返回值”的相关文章。那么小编也在网上搜集了一些关于“nginx打印lua返回值””的相关内容,希望我们能喜欢,小伙伴们快快来学习一下吧!

限流是一个很大的话题,准备把其中的所有限流器都实现一遍,以此也算全都写过了,到时候再用也不至于会心虚,毕竟真实写完成过。本文主要讲述了如何基于 Redis 与 Lua实现分布式令牌桶的限流方案。

读前提问

我觉得学习任何东西前都应该有自己的反问,这种反问基于标题给你的大概印象。带着问题来看文章,最后应该比盲目的看有收获,先提出几个基础的问题。

限流是什么

通过某种手段对某个时间段的并发访问请求进行流量限制,一旦流量达到限制阈值则可以拒绝服务,排队或等待,目的是防止系统因大流量或突发流量导致服务不可用或崩溃,是一种确保系统高可用的手段。

限流的简单了解

限流常见场景

对外限流:

电商秒杀(因秒杀业务特性,需要限流):到达开卖时间瞬间大流量,此时下单人数>商品库存,服务器不可能同时全部消费,需要进行限流,卖完了之后就拒绝后续下单请求。 微博热搜(因产品特性,需要限流):突然出现了几个大瓜,那微博是不是突然流量激增,重灾区就是微博热搜,此时所有服务满载运行,必须有个限流策略保证服务的高可用。 防止恶意攻击(突发恶意攻击,需要限流):比如某一个 API 被疯狂请求,或者某一个 IP 疯狂请求公司的 API,此时就需要进行限流,常见措施是先告警,再限流。为了不影响其他服务的正常使用,需要设计限流方案。 API有偿调用:用户认证+限流策略,顾名思义没啥好说的,一般是 SAAS 公司最常见的业务,常见于 OPEN-API 相关的小组负责的。

对内限流:

BUG预防:核心服务的高可用是十分重要的,千万不能挂。如果内部应用出现 bug,一直调用核心服务,核心服务就有被击垮的风险,限流也十分重要。 缓存雪崩:请求直接打到 DB,那就哦豁完蛋了,所以需要根据业务场景来实现限流后是排队还是丢弃。

综上所述,需要进行限流的场景可以分为三种:

公共的 API ,限流策略用于open-api 网关与相关服务的可用性,同时可以防止恶意攻击。内部的核心应用,应对 bug 或其他突发情况,目的就是保证突发情况下核心应用的高可用。产品具备突发大流量请求的特性,妥妥的都给加上限流策略,保证整个系统的高可用。

限流解决了什么问题

保证服务高可用,牺牲一部分的流量,换取服务的可用性。对于被限流器直接作用的应用来说,除了保证自身不被流量击垮,还保护了依赖它的下游应用。

限流带来的问题

任何技术都是双刃剑,没有绝对的好用,能带来优点必然也会带来问题。

限流组件保证了高可用,牺牲了性能,增加了一层 IO 环节的开销,单机限流在本地,分布式限流还要通过网络协议。限流组件保证了高可用,牺牲了一致性,在大流量的情况下,请求的处理会出现延迟的情况,这种场景便无法保证强一致性。特殊情况下,还无法保证最终一致性,部分请求直接被抛弃。限流组件拥有流控权,若限流组件挂了,会引起雪崩效应,导致请求与业务的大批量失败。引入限流组件,增加系统的复杂程度,开发难度增加,限流中间件的设计本身就是一个复杂的体系,需要综合业务与技术去思考与权衡,同时还要确保限流组件本身的高可用与性能,极大增加工作量,甚至需要一个团队去专门开发。

设计限流组件本身需要考虑的点

如果我来设计限流组件,我大致会确认如下几个点:

1.明确限流器的目的:

用在哪些模块? 应对哪些场景下的什么问题? 是单机限流还是分布式限流? 确定限流模块的使用层面?例如:单应用维度、业务域维度、网关维度

2.明确限流器的维度,例如 IP 维度,用户授权 token 维度,API 维度等

3.怎么保证限流组件的高可用?

4.怎么解决使用限流组件后带来的一致性问题?

5.怎么缩小限流器的粒度,实现平滑限流?

常见的限流实现

单机

基于Java 并发工具

(信号量 / concurrentHashMap)

基于Google Guava RateLimiter

稳定模式(SmoothBursty:令牌生成速度恒定) / 渐进模式(SmoothWarmingUp:令牌生成速度缓慢提升直到维持在一个稳定值)

分布式

(Redis + Lua / Nginx + Lua)

常见限流器种类

这四种限流器虽然网上介绍得很多,但是我写给自己看的 ^_^,自己要每次遇到都能够脱口而出,而不是“我经常看到过,但是我记不起来了”或者“我知道是什么意思,但是我就是说不出来,也说不清楚”。后续, 等API网关的限流模块代码完成后, 对着代码和实践会仔细展开说说 ~

计数器(固定窗口限流器)滑动窗口限流器令牌桶限流器漏桶限流器开始实践

模拟的场景

模拟API 网关中的一个 API 接口在某个时刻突然接收到 100 个并发请求,但是该 API 配置的令牌桶限流器每1分钟生成一个,每次限流间隔为 1 小时,限流上限为 60,则通过代码模拟出最终效果,并输出日志。

实现的效果

构建请求

通过参数可知,限流器的类别LimiterType选择的是令牌桶,限流的时间单位timeUnit是每小时,每个限流时间内的令牌桶内令牌的最大数量value是 60.

{    "id": 3,    "apiId": 3,    "apiName": "测试API",    "ip": "127.0.0.1",    "dimensionName": "app_id",    "dimensionValue": "testid1234",    "timeUnit": 2,    "value": 60,    "LimiterType": 1}

使用 PostMan 中的迭代器功能,进行循环请求:

计算令牌桶与推测

限流间隔是 1 小时桶内最大令牌是 60 个计算得出令牌的生成间隔是 1 个/1 分钟模拟并发请求 100 个,每个请求的间隔时间是 0ms此时令牌并未来得及生成令牌,所以在第 61 个并发的时候请求,令牌用光被限流

请求的结果

通过下图可知与上面推测相符合,第 61 个请求被限流。

关键代码

总的来说,这个模块的流程比较简单,所以直接理解关键代码就 ok 了,实现起来也很容易。

限流器的抽象设计:

预计实现四种限流器,目前本文内实现的是令牌桶限流器。限流器的抽象设计是经典的三层结构,也采用了模板方法的思想,也就是最上层的接口,实现一些公共方法与公共抽象的顶层抽象类,最后是每个限流器的独有逻辑放在各自类中来做。

限流业务的实现:

这里贴出限流业务的核心方法,通过调用doFilter 方法实现判断是否需要进行限流。具体调用哪一种限流器通过这两个对象实现的:LimiterStrategy 与 LimiterStrategy 分别是具体的限流算法与限流策略。

@Override    public boolean doFilter(FlowControlConfig flowControlConfig) {        if (Objects.isNull(flowControlConfig)) {            log.error("[{}] 流控参数为空", this.getClass().getSimpleName());            return true;        }        String key;        boolean filterRes = true;        try {            key = generateRedisLimiterKey(flowControlConfig);            LimiterStrategy limiterStrategy = getLimiterStrategyByCode(flowControlConfig.getLimiterType());            LimiterPolicy limiterPolicy = getLimiterPolicyByCode(flowControlConfig.getLimiterType(), flowControlConfig);            filterRes = limiterStrategy.access(key, limiterPolicy);            if (!filterRes) {                log.warn("Limiter Id:[{}],key :[{}]已达流量上限值:{},被限制请求!", flowControlConfig.getId(), key, flowControlConfig.getValue());                // todo 接入消息告警            }        } catch (Exception e) {            log.error("[{}] 限流器内部出现异常! 入参:{}", this.getClass().getSimpleName(), JSONObject.toJSON(flowControlConfig));            e.printStackTrace();        }        return !filterRes;    }

令牌桶限流器算法的对象:

package com.teavamc.rpcgateway.core.flow.limiter.policy;import com.google.common.collect.Lists;import java.util.List;/** * 令牌桶限流器的执行对象 * * @Package com.teavamc.rpcgateway.core.limiter.policy * @date 2021/1/28 上午11:09 */public class TokenBucketLimiterPolicy extends AbstractLimiterPolicy {    /**     * 限流时间间隔     * (重置桶内令牌的时间间隔)     */    private final long resetBucketInterval;    /**     * 最大令牌数量     */    private final long bucketMaxTokens;    /**     * 初始可存储数量     */    private final long initTokens;    /**     * 每个令牌产生的时间     */    private final long intervalPerPermit;    /**     * 令牌桶对象的构造器     * @param bucketMaxTokens 桶的令牌上限     * @param resetBucketInterval 限流时间间隔     * @param maxBurstTime 最大的突发流量的持续时间(通过计算)     */    public TokenBucketLimiterPolicy(long bucketMaxTokens, long resetBucketInterval, long maxBurstTime) {        // 最大令牌数        this.bucketMaxTokens = bucketMaxTokens;        // 限流时间间隔        this.resetBucketInterval = resetBucketInterval;        // 令牌的产生间隔 = 限流时间 / 最大令牌数        intervalPerPermit = resetBucketInterval / bucketMaxTokens;        // 初始令牌数 = 最大的突发流量的持续时间 / 令牌产生间隔        // 用 最大的突发流量的持续时间 计算的结果更加合理,并不是每次初始化都要将桶装满        initTokens = Math.min(maxBurstTime / intervalPerPermit, bucketMaxTokens);    }    public long getResetBucketInterval() {        return resetBucketInterval;    }    public long getBucketMaxTokens() {        return bucketMaxTokens;    }    public long getInitTokens() {        return initTokens;    }    public long getIntervalPerPermit() {        return intervalPerPermit;    }    @Override    public String[] toParams() {        List<String > list = Lists.newArrayList();        list.add(String.valueOf(getIntervalPerPermit()));        list.add(String.valueOf(System.currentTimeMillis()));        list.add(String.valueOf(getInitTokens()));        list.add(String.valueOf(getBucketMaxTokens()));        list.add(String.valueOf(getResetBucketInterval()));        return list.toArray(new String[]{});    }}

这个代码已经写得很明白了,东西也不多。但是构造器这里还是要理解一下,特别是maxBurstTime 这个字段,记录这个 api 经历的最大突发流量的时间。

Lua 脚本的解析:

令牌桶的实现是通过 lua 来完成的,所以 lua 是核心逻辑。这是我这边使用的令牌桶方案,都加了注解,如果看不懂就多看几遍,还是看不明白就看最后我的流程图。

--[[  1. key - 令牌桶的 key  2. intervalPerTokens - 生成令牌的间隔(ms)  3. curTime - 当前时间  4. initTokens - 令牌桶初始化的令牌数  5. bucketMaxTokens - 令牌桶的上限  6. resetBucketInterval - 重置桶内令牌的时间间隔  7. currentTokens - 当前桶内令牌数  8. bucket - 当前 key 的令牌桶对象]] --local key = KEYS[1]local intervalPerTokens = tonumber(ARGV[1])local curTime = tonumber(ARGV[2])local initTokens = tonumber(ARGV[3])local bucketMaxTokens = tonumber(ARGV[4])local resetBucketInterval = tonumber(ARGV[5])local bucket = redis.call('hgetall', key)local currentTokens-- 若当前桶未初始化,先初始化令牌桶if table.maxn(bucket) == 0 then    -- 初始桶内令牌    currentTokens = initTokens    -- 设置桶最近的填充时间是当前    redis.call('hset', key, 'lastRefillTime', curTime)    -- 初始化令牌桶的过期时间, 设置为间隔的 1.5 倍    redis.call('pexpire', key, resetBucketInterval * 1.5)-- 若桶已初始化,开始计算桶内令牌-- 为什么等于 4 ? 因为有两对 field, 加起来长度是 4-- { "lastRefillTime(上一次更新时间)","curTime(更新时间值)","tokensRemaining(当前保留的令牌)","令牌数" }elseif table.maxn(bucket) == 4 then    -- 上次填充时间    local lastRefillTime = tonumber(bucket[2])    -- 剩余的令牌数    local tokensRemaining = tonumber(bucket[4])    -- 当前时间大于上次填充时间    if curTime > lastRefillTime then        -- 拿到当前时间与上次填充时间的时间间隔        -- 举例理解: curTime = 2620 , lastRefillTime = 2000, intervalSinceLast = 620        local intervalSinceLast = curTime - lastRefillTime        -- 如果当前时间间隔 大于 令牌的生成间隔        -- 举例理解: intervalSinceLast = 620, resetBucketInterval = 1000        if intervalSinceLast > resetBucketInterval then            -- 将当前令牌填充满            currentTokens = initTokens            -- 更新重新填充时间            redis.call('hset', key, 'lastRefillTime', curTime)                    -- 如果当前时间间隔 小于 令牌的生成间隔        else            -- 可授予的令牌 = 向下取整数( 上次填充时间与当前时间的时间间隔 / 两个令牌许可之间的时间间隔 )            -- 举例理解 : intervalPerTokens = 200 ms , 令牌间隔时间为 200ms            --           intervalSinceLast = 620 ms , 当前距离上一个填充时间差为 620ms            --           grantedTokens = 620/200 = 3.1 = 3            local grantedTokens = math.floor(intervalSinceLast / intervalPerTokens)            -- 可授予的令牌 > 0 时            -- 举例理解 : grantedTokens = 620/200 = 3.1 = 3            if grantedTokens > 0 then                -- 生成的令牌 = 上次填充时间与当前时间的时间间隔 % 两个令牌许可之间的时间间隔                -- 举例理解 : padMillis = 620%200 = 20                --           curTime = 2620                --           curTime - padMillis = 2600                local padMillis = math.fmod(intervalSinceLast, intervalPerTokens)                -- 将当前令牌桶更新到上一次生成时间                redis.call('hset', key, 'lastRefillTime', curTime - padMillis)            end            -- 更新当前令牌桶中的令牌数            -- Math.min(根据时间生成的令牌数 + 剩下的令牌数, 桶的限制) => 超出桶最大令牌的就丢弃            currentTokens = math.min(grantedTokens + tokensRemaining, bucketMaxTokens)        end    else        -- 如果当前时间小于或等于上次更新的时间, 说明刚刚初始化, 当前令牌数量等于桶内令牌数        -- 不需要重新填充        currentTokens = tokensRemaining    endend-- 如果当前桶内令牌小于 0,抛出异常assert(currentTokens >= 0)-- 如果当前令牌 == 0 ,更新桶内令牌, 返回 0if currentTokens == 0 then    redis.call('hset', key, 'tokensRemaining', currentTokens)    return 0else    -- 如果当前令牌 大于 0, 更新当前桶内的令牌 -1 , 再返回当前桶内令牌数    redis.call('hset', key, 'tokensRemaining', currentTokens - 1)    return currentTokensend

其实这个脚本很简单,一个 key 拥有一个令牌桶,令牌桶是通过 Redis 中的 Hash 数据类型进行储存的。每个令牌桶拥有两个 field,分别是上一次填充时间lastRefillTime与当前桶内令牌数量tokensRemaining。

从脚本逻辑上来说,就分成了三个步骤,分别是:

1.确认 key 的令牌桶是否存在,如果不存在就初始化。

2.计算并更新当前令牌桶内的令牌数量:

如果当前距离上次填充令牌的时间间隔超出重置时间,就重置令牌桶。 计算距离上次填充的时间间隔是否超过了生产令牌的间隔时间,若大于间隔就计算生产了多少令牌与上次产生令牌的时间。 若距离上次填充至今没有产生令牌就直接用。

3.明确了当前桶内的令牌数之后,就判断是否放行:

令牌等于 0,返回 0,不放行。 令牌大于0,减少一个当前的桶内令牌,放行。

限流器的模拟使用:

开启一个接口,模拟对接口并发调用。

@PostMapping(value = "/test")    public void testFlowControl(@RequestBody FlowControlConfig controlConfig) {        Long apiId = controlConfig.getId();        log.info("接收到 ApiId :{} 的请求", apiId);        apiRequestCount.put(apiId, apiRequestCount.getOrDefault(apiId, 0) + 1);        // 执行限流        boolean res = flowControl.doFilter(controlConfig);        if (res) {            apiRequestFailedCount.put(apiId, apiRequestFailedCount.getOrDefault(apiId, 0) + 1);        } else {            apiRequestSuccessCount.put(apiId, apiRequestSuccessCount.getOrDefault(apiId, 0) + 1);        }        // 处理结果        int totalCnt = apiRequestCount.get(apiId);        int successCnt = apiRequestSuccessCount.get(apiId) == null ? 0 : apiRequestSuccessCount.get(apiId);        int failedCnt = apiRequestFailedCount.get(apiId) == null ? 0 : apiRequestFailedCount.get(apiId);        log.info(" ApiId :{} 的请求是否被限流:{} | 共请求{}次,放行{}次,限流{}次", apiId, res, totalCnt, successCnt, failedCnt);    }

后续业务拓展需要考虑的点:

弹性限流怎么做?平滑限流怎么做?关于api网关的调用的耗时比的思考?网关的性能计算,怎么计算 qps,怎么计算怎么抗多少?怎么合理估算API 的性能,并设置合适的限流大小?怎么根据业务场景选择合适的限流方案?最后结尾

第二篇文章,其实是写给自己看的,帮助自己查漏补缺。写成公开文章总要更加仔细,所以其实怎么看都是有益的,除非就是很费头发罢了。这篇文章帮助我自己实现了 API 网关 DEMO中限流模块的一部分,我只是将分布式令牌桶的实现拿出来做成了一篇文章,限流模块的设计后续会分享。

代码地址

有兴趣的可以看看,因为是上班抽空写的,还不是很完善~

Github:github.com/teavmac/jav…

作者:teavmac

链接:

来源:掘金

标签: #nginx打印lua返回值