rpc-tech-stack 系列的实践文章 ~ 本文属于限流话题.

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

读前提问

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

限流是什么

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

限流的简单了解

限流常见场景

对外限流：
- 电商秒杀（因秒杀业务特性，需要限流）：到达开卖时间瞬间大流量，此时下单人数>商品库存，服务器不可能同时全部消费，需要进行限流，卖完了之后就拒绝后续下单请求。
- 微博热搜（因产品特性，需要限流）：突然出现了几个大瓜，那微博是不是突然流量激增，重灾区就是微博热搜，此时所有服务满载运行，必须有个限流策略保证服务的高可用。
- 防止恶意攻击（突发恶意攻击，需要限流）：比如某一个 API 被疯狂请求，或者某一个 IP 疯狂请求公司的 API，此时就需要进行限流，常见措施是先告警，再限流。为了不影响其他服务的正常使用，需要设计限流方案。
- API有偿调用：用户认证+限流策略，顾名思义没啥好说的，一般是 SAAS 公司最常见的业务，常见于 OPEN-API 相关的小组负责的。
对内限流
- BUG预防：核心服务的高可用是十分重要的，千万不能挂。如果内部应用出现 bug，一直调用核心服务，核心服务就有被击垮的风险，限流也十分重要。
- 缓存雪崩：请求直接打到 DB，那就哦豁完蛋了，所以需要根据业务场景来实现限流后是排队还是丢弃。

综上所得，需要进行限流的场景可以分为三种：

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

限流解决了什么问题

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

限流带来的问题

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

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

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

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

明确限流器的目的：
- 用在哪些模块？
- 应对哪些场景下的什么问题?
- 是单机限流还是分布式限流?
- 确定限流模块的使用层面？例如：单应用维度、业务域维度、网关维度
明确限流器的维度，例如 IP 维度，用户授权 token 维度，API 维度等
怎么保证限流组件的高可用？
怎么解决使用限流组件后带来的一致性问题？
怎么缩小限流器的粒度，实现平滑限流？

常见的限流实现

单机
- 基于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 分别是具体的限流算法与限流策略。

@Overridepublic 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;}@Overridepublic 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 - 令牌桶的 key2. 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 = 620local intervalSinceLast = curTime - lastRefillTime-- 如果当前时间间隔 大于 令牌的生成间隔-- 举例理解: intervalSinceLast = 620, resetBucketInterval = 1000if 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 = 3local grantedTokens = math.floor(intervalSinceLast / intervalPerTokens)-- 可授予的令牌 > 0 时-- 举例理解 : grantedTokens = 620/200 = 3.1 = 3if grantedTokens > 0 then-- 生成的令牌 = 上次填充时间与当前时间的时间间隔 % 两个令牌许可之间的时间间隔-- 举例理解 : padMillis = 620%200 = 20--           curTime = 2620--           curTime - padMillis = 2600local padMillis = math.fmod(intervalSinceLast, intervalPerTokens)-- 将当前令牌桶更新到上一次生成时间redis.call('hset', key, 'lastRefillTime', curTime - padMillis)end-- 更新当前令牌桶中的令牌数-- Math.min(根据时间生成的令牌数 + 剩下的令牌数, 桶的限制) => 超出桶最大令牌的就丢弃currentTokens = math.min(grantedTokens + tokensRemaining, bucketMaxTokens)endelse-- 如果当前时间小于或等于上次更新的时间, 说明刚刚初始化, 当前令牌数量等于桶内令牌数-- 不需要重新填充currentTokens = tokensRemainingend
end-- 如果当前桶内令牌小于 0,抛出异常
assert(currentTokens >= 0)-- 如果当前令牌 == 0 ,更新桶内令牌, 返回 0
if currentTokens == 0 thenredis.call('hset', key, 'tokensRemaining', currentTokens)return 0
else-- 如果当前令牌 大于 0, 更新当前桶内的令牌 -1 , 再返回当前桶内令牌数redis.call('hset', key, 'tokensRemaining', currentTokens - 1)return currentTokens
end

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

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

确认 key 的令牌桶是否存在，如果不存在就初始化。
计算并更新当前令牌桶内的令牌数量：
- 如果当前距离上次填充令牌的时间间隔超出重置时间，就重置令牌桶。
- 计算距离上次填充的时间间隔是否超过了生产令牌的间隔时间，若大于间隔就计算生产了多少令牌与上次产生令牌的时间。
- 若距离上次填充至今没有产生令牌就直接用。
明确了当前桶内的令牌数之后，就判断是否放行：
- 令牌等于 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：https://github.com/teavmac/java-rpc-tech-stack

基于Redis和 Lua 实现分布式令牌桶限流相关推荐

可能要用心学高并发核心编程，限流原理与实战，分布式令牌桶限流
实战:分布式令牌桶限流本节介绍的分布式令牌桶限流通过Lua+Java结合完成,首先在Lua脚本中完成限流的计算,然后在Java代码中进行组织和调用. 分布式令牌桶限流Lua脚本分布式令牌桶限流Lu ...
Jedis使用lua脚本完成令牌桶限流
Jedis使用lua脚本完成令牌桶限流文章目录 Jedis使用lua脚本完成令牌桶限流一.lua脚本的简单语法二.令牌桶限流 1. 构思 2. 实现三.Jemeter压测工具测试一.lua脚 ...
ASP.NET Core中使用令牌桶限流
在限流时一般会限制每秒或每分钟的请求数,简单点一般会采用计数器算法,这种算法实现相对简单,也很高效,但是无法应对瞬时的突发流量. 比如限流每秒100次请求,绝大多数的时间里都不会超过这个数,但是偶尔某 ...
【秒杀系统】零基础上手秒杀系统（二）：令牌桶限流 + 再谈超卖
前言本文是秒杀系统的第二篇,通过实际代码讲解,帮助你快速的了解秒杀系统的关键点,上手实际项目. 本篇主要讲解接口限流措施,接口限流其实定义也非常广,接口限流本身也是系统安全防护的一种措施,暂时列举这 ...
什么是限流？为什么会限流呢？常见的限流算法【固定窗口限流、滑动窗口限流、漏桶限流、令牌桶限流】是什么呢？
什么是限流?为什么会限流呢?常见的限流算法[固定窗口限流.滑动窗口限流.漏桶限流.令牌桶限流]是什么呢? 什么是限流? 为什么会限流? 1. 固定窗口限流算法 1.1 什么是固定窗口限流算法 1.2 ...
十三水算法php_基于PHP+Redis令牌桶限流
一 .场景描述在开发接口服务器的过程中,为了防止客户端对于接口的滥用,保护服务器的资源, 通常来说我们会对于服务器上的各种接口进行调用次数的限制.比如对于某个用户,他在一个时间段(interval ...
分布式限流实战--redis实现令牌桶限流
这篇文章我们主要是分析一下分布式限流的玩法. 因为限流也是一个经典用法了. 1.微服务限流随着微服务的流行,服务和服务之间的稳定性变得越来越重要.缓存.降级和限流是保护微服务系统运行稳定性的三大利器 ...
php限制接口访问次数_令牌桶限流思路分享（PHP+Redis实现机制）
一 .场景描述在开发接口服务器的过程中,为了防止客户端对于接口的滥用,保护服务器的资源, 通常来说我们会对于服务器上的各种接口进行调用次数的限制.比如对于某个用户,他在一个时间段(interval ...
rateLimiter令牌桶限流算法
RateLimiter是guava提供的基于令牌桶算法的实现类,可以非常简单的完成限流特技,并且根据系统的实际情况来调整生成token的速率. 通常可应用于抢购限流防止冲垮系统:限制某接口.服务单位时 ...

基于Redis和 Lua 实现分布式令牌桶限流