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 中的迭代器功能,进行循环请求:

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

计算令牌桶与推测

  • 限流间隔是 1 小时

  • 桶内最大令牌是 60 个

  • 计算得出令牌的生成间隔是 1 个/1 分钟

  • 模拟并发请求 100 个,每个请求的间隔时间是 0ms

  • 此时令牌并未来得及生成令牌,所以在第 61 个并发的时候请求,令牌用光被限流

请求的结果

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

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

关键代码

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

限流器的抽象设计:

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

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

限流业务的实现:

这里贴出限流业务的核心方法,通过调用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 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)

自我介绍一下,小编13年上海交大毕业,曾经在小公司待过,也去过华为、OPPO等大厂,18年进入阿里一直到现在。

深知大多数Java工程师,想要提升技能,往往是自己摸索成长或者是报班学习,但对于培训机构动则几千的学费,着实压力不小。自己不成体系的自学效果低效又漫长,而且极易碰到天花板技术停滞不前!

因此收集整理了一份《2024年Java开发全套学习资料》,初衷也很简单,就是希望能够帮助到想自学提升又不知道该从何学起的朋友,同时减轻大家的负担。img

既有适合小白学习的零基础资料,也有适合3年以上经验的小伙伴深入学习提升的进阶课程,基本涵盖了95%以上Java开发知识点,真正体系化!

由于文件比较大,这里只是将部分目录截图出来,每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频,并且会持续更新!

如果你觉得这些内容对你有帮助,可以扫码获取!!(备注Java获取)

img

总结

大型分布式系统犹如一个生命,系统中各个服务犹如骨骼,其中的数据犹如血液,而Kafka犹如经络,串联整个系统。这份Kafka源码笔记通过大量的设计图展示、代码分析、示例分享,把Kafka的实现脉络展示在读者面前,帮助读者更好地研读Kafka代码。

麻烦帮忙转发一下这篇文章+关注我

就这一次!拼多多内部架构师培训Kafka源码笔记(现已绝版)

《互联网大厂面试真题解析、进阶开发核心学习笔记、全套讲解视频、实战项目源码讲义》点击传送门即可获取!
基础资料,也有适合3年以上经验的小伙伴深入学习提升的进阶课程,基本涵盖了95%以上Java开发知识点,真正体系化!**

由于文件比较大,这里只是将部分目录截图出来,每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频,并且会持续更新!

如果你觉得这些内容对你有帮助,可以扫码获取!!(备注Java获取)

img

总结

大型分布式系统犹如一个生命,系统中各个服务犹如骨骼,其中的数据犹如血液,而Kafka犹如经络,串联整个系统。这份Kafka源码笔记通过大量的设计图展示、代码分析、示例分享,把Kafka的实现脉络展示在读者面前,帮助读者更好地研读Kafka代码。

麻烦帮忙转发一下这篇文章+关注我

[外链图片转存中…(img-Uv0CaVlU-1713581011109)]

《互联网大厂面试真题解析、进阶开发核心学习笔记、全套讲解视频、实战项目源码讲义》点击传送门即可获取!

Logo

北京人形旗下天工造物具身智能开源社区,聚焦具身天工与慧思开物两大平台

更多推荐