redisson分布式限流RRateLimiter源碼解析
更新時(shí)間：2022年11月01日 16:15:56 作者：你行你上啊
這篇文章主要為大家介紹了redisson分布式限流RRateLimiter源碼解析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪
分布式限流-單位時(shí)間多實(shí)例多線程訪問(wèn)次數(shù)限制
接前面聊一聊redisson及優(yōu)雅實(shí)現(xiàn) 和 說(shuō)一說(shuō)spring boot優(yōu)雅集成redisson，簡(jiǎn)單以源碼的方式給大家介紹了redisson的：可重入性、阻塞、續(xù)約、紅鎖、聯(lián)鎖、加鎖解鎖流程和集成spring boot注意點(diǎn)和優(yōu)雅實(shí)現(xiàn)方式。
接下來(lái)在講一講平時(shí)用的比較多的限流模塊--RRateLimiter
1.簡(jiǎn)單使用
public static void main(String[] args) throws InterruptedException {
RRateLimiter rateLimiter = createLimiter();
int allThreadNum = 20;
CountDownLatch latch = new CountDownLatch(allThreadNum);
long startTime = System.currentTimeMillis();
for (int i = 0; i < allThreadNum; i++) {
// new Thread(() -> {
if(i % 3 == 0) Thread.sleep(1000);
boolean pass = rateLimiter.tryAcquire();
if(pass) {
log.info("get ");
} else {
log.info("no");
}
// latch.countDown();
// }).start();
}
// latch.await();
System.out.println("Elapsed " + (System.currentTimeMillis() - startTime));
}
public static RRateLimiter createLimiter() {
Config config = new Config();
config.useSingleServer()
.setTimeout(1000000)
.setPassword("123456")
.setAddress("redis://xxxx:6379");
RedissonClient redisson = Redisson.create(config);
RRateLimiter rateLimiter = redisson.getRateLimiter("myRateLimiter3");
// 初始化：PER_CLIENT 單實(shí)例執(zhí)行，OVERALL 全實(shí)例執(zhí)行
// 最大流速 = 每10秒鐘產(chǎn)生3個(gè)令牌
rateLimiter.trySetRate(RateType.OVERALL, 3, 10, RateIntervalUnit.SECONDS);
return rateLimiter;
}
實(shí)際結(jié)果：
[2022-10-29 14:32:46.261][INFO ][main][][] RedisTest - get
[2022-10-29 14:32:46.312][INFO ][main][][] RedisTest - get
[2022-10-29 14:32:46.358][INFO ][main][][] RedisTest - get
[2022-10-29 14:32:47.416][INFO ][main][][] RedisTest - no
[2022-10-29 14:32:47.469][INFO ][main][][] RedisTest - no
[2022-10-29 14:32:47.517][INFO ][main][][] RedisTest - no
[2022-10-29 14:32:48.577][INFO ][main][][] RedisTest - no
[2022-10-29 14:32:48.623][INFO ][main][][] RedisTest - no
2. 實(shí)現(xiàn)限流redisson使用了哪些redis數(shù)據(jù)結(jié)構(gòu)
- Hash結(jié)構(gòu) -- 限流器結(jié)構(gòu)：
參數(shù)rate代表速率
參數(shù)interval代表多少時(shí)間內(nèi)產(chǎn)生的令牌
參數(shù)type代表單機(jī)還是集群
- ZSET結(jié)構(gòu) -- 記錄獲取令牌的時(shí)間戳，用于時(shí)間對(duì)比。
1667025166312 --> 2022-10-29 14:32:46
1667025166262 --> 2022-10-29 14:32:46
1667025166215 --> 2022-10-29 14:32:46
- String結(jié)構(gòu) --記錄的是當(dāng)前令牌桶中的令牌數(shù)【很明顯被我用完了現(xiàn)在是0】
3. 超過(guò)10s,我再次獲取一個(gè)令牌，數(shù)據(jù)結(jié)構(gòu)發(fā)生的變化
- ZSET結(jié)構(gòu)。-- 新生成一個(gè)ZSET結(jié)構(gòu)，存放獲取令牌的時(shí)間戳
- String 結(jié)構(gòu) --當(dāng)前令牌桶還有2個(gè)令牌
4. 源碼淺析
RRateLimiter rateLimiter = redisson.getRateLimiter("myRateLimiter3");
// 初始化
// 最大流速 = 每10秒鐘產(chǎn)生3個(gè)令牌
rateLimiter.trySetRate(RateType.PER_CLIENT, 3, 10, RateIntervalUnit.SECONDS);
初始化定義沒(méi)有什么好講的，就是創(chuàng)建HASH結(jié)構(gòu)
主要還是講講： rateLimiter.tryAcquire()
private <T> RFuture<T> tryAcquireAsync(RedisCommand<T> command, Long value) {
return this.commandExecutor.evalWriteAsync(
this.getName(), LongCodec.INSTANCE, command,
"local rate = redis.call('hget', KEYS[1], 'rate');
local interval = redis.call('hget', KEYS[1], 'interval');
local type = redis.call('hget', KEYS[1], 'type');
assert(
rate ~= false and interval ~= false and type ~= false, 'RateLimiter is not initialized'
)
local valueName = KEYS[2];
local permitsName = KEYS[4];
if type == '1' then valueName = KEYS[3];
permitsName = KEYS[5];
end;
local currentValue = redis.call('get', valueName);
if currentValue ~= false
then
local expiredValues = redis.call(
'zrangebyscore', permitsName, 0, tonumber(ARGV[2]
)
- interval
);
local released = 0;
for i, v in ipairs(expiredValues)
do local random, permits = struct.unpack('fI', v);
released = released + permits;end;
if released > 0
then
redis.call('zrem', permitsName, unpack(expiredValues));
currentValue = tonumber(currentValue) + released;
redis.call('set', valueName, currentValue);
end;
if tonumber(currentValue) < tonumber(ARGV[1])
then
local nearest = redis.call(
'zrangebyscore', permitsName, '(' .. (tonumber(ARGV[2])
- interval), tonumber(ARGV[2]), 'withscores', 'limit', 0, 1);
local random, permits = struct.unpack('fI', nearest[1]);
return tonumber(nearest[2])
- (tonumber(ARGV[2])
- interval);
else
redis.call('zadd', permitsName, ARGV[2], struct.pack('fI', ARGV[3], ARGV[1]));
redis.call('decrby', valueName, ARGV[1]);
return nil;
end;
else
assert(tonumber(rate) >= tonumber(ARGV[1]), 'Requested permits amount could not exceed defined rate');
redis.call('set', valueName, rate);
redis.call('zadd', permitsName, ARGV[2], struct.pack('fI', ARGV[3], ARGV[1]));
redis.call('decrby', valueName, ARGV[1]);
return nil;
end;", Arrays.asList(this.getName(),
this.getValueName(),
this.getClientValueName(),
this.getPermitsName(),
this.getClientPermitsName()),
new Object[]{value,
System.currentTimeMillis(),
ThreadLocalRandom.current().nextLong()
}
);
}
主要就是這段lua代碼，下面我詳細(xì)過(guò)一下
作者目前用的3.16.3版本，剛好遇見(jiàn)redisson的bug，見(jiàn)3197，請(qǐng)大家用最新版本，以下為修復(fù)后解析。
-- 獲取hash結(jié)構(gòu)的速率
local rate = redis.call("hget", KEYS[1], "rate")
-- 獲取hash結(jié)構(gòu)的時(shí)間區(qū)間(ms)
local interval = redis.call("hget", KEYS[1], "interval")
-- 獲取hash結(jié)構(gòu)的時(shí)間類型
local type = redis.call("hget", KEYS[1], "type")
-- 判斷是否初始化限流結(jié)構(gòu)
assert(rate ~= false and interval ~= false and type ~= false, "RateLimiter is not initialized")
-- {name}:value string結(jié)構(gòu)，這個(gè)key記錄的是當(dāng)前令牌桶中的令牌數(shù)
local valueName = KEYS[2]
-- {name}:permits zset結(jié)構(gòu)，記錄了請(qǐng)求的令牌數(shù)，score則為請(qǐng)求的時(shí)間戳
local permitsName = KEYS[4]
-- 單機(jī)限流才會(huì)用到，集群模式不用關(guān)注
if type == "1" then
valueName = KEYS[3]
permitsName = KEYS[5]
end
-- 生產(chǎn)速率rate必須比請(qǐng)求的令牌數(shù)大
assert(tonumber(rate) >= tonumber(ARGV[1]), "Requested permits amount could not exceed defined rate")
-- 初始化RateLimiter并不會(huì)初始化stirng結(jié)構(gòu)，因此第一次獲取這里currentValue是null
local currentValue = redis.call("get", valueName)
if currentValue ~= false then
-- 第二次獲取令牌執(zhí)行
-------------------------- 獲取zset結(jié)構(gòu)：統(tǒng)計(jì)之前的請(qǐng)求令牌數(shù)
-- 范圍是0 ~ (第二次請(qǐng)求時(shí)間戳 - 令牌生產(chǎn)的時(shí)間)
local expiredValues = redis.call("zrangebyscore", permitsName, 0, tonumber(ARGV[2]) - interval)
local released = 0
-- lua迭代器，遍歷expiredValues，如果有值，那么released等于之前所有請(qǐng)求的令牌數(shù)之和，表示應(yīng)該釋放多少令牌
for i, v in ipairs(expiredValues) do
-- 獲取請(qǐng)求數(shù)permits
local random, permits = struct.unpack("fI", v)
released = released + permits
end
-- 之前的請(qǐng)求令牌數(shù) > 0, 例如10s產(chǎn)生3個(gè)令牌，現(xiàn)在超過(guò)10s了，重置周期并計(jì)算剩余令牌數(shù)
if released > 0 then
-- 移除zset中所有元素【要求是同一個(gè)限流器permitsName，不然就移除不了，尷尬】
redis.call("zrem", permitsName, unpack(expiredValues))
currentValue = tonumber(currentValue) + released
------------------------- 更新string結(jié)構(gòu)：=剩下令牌數(shù)+釋放令牌數(shù)
redis.call("set", valueName, currentValue)
end
-- 如果當(dāng)前令牌數(shù) ＜ 請(qǐng)求的令牌數(shù)
if tonumber(currentValue) < tonumber(ARGV[1]) then
-- 從zset中找到距離當(dāng)前時(shí)間最近的那個(gè)請(qǐng)求，也就是上一次放進(jìn)去的請(qǐng)求信息
local nearest = redis.call('zrangebyscore', permitsName, '(' .. (tonumber(ARGV[2]) - interval), tonumber(ARGV[2]), 'withscores', 'limit', 0, 1);
local random, permits = struct.unpack("fI", nearest[1])
-- 返回 上一次請(qǐng)求的時(shí)間戳 - (當(dāng)前時(shí)間戳 - 令牌生成的時(shí)間間隔) 這個(gè)值表示還需要多久才能生產(chǎn)出足夠的令牌
return tonumber(nearest[2]) - (tonumber(ARGV[2]) - interval)
else
-- 如果當(dāng)前令牌數(shù) ≥ 請(qǐng)求的令牌數(shù)，表示令牌夠多，更新zset
------------------------- 更新zset結(jié)構(gòu)
redis.call("zadd", permitsName, ARGV[2], struct.pack("fI", ARGV[3], ARGV[1]))
------------------------- 更新Stringt結(jié)構(gòu)，減少一個(gè)剩下的令牌數(shù)
redis.call("decrby", valueName, ARGV[1])
return nil
end
else
--------汀雨筆記----------------- 初始化Stringt結(jié)構(gòu)，當(dāng)前限流器的令牌數(shù)
redis.call("set", valueName, rate)
--------汀雨筆記----------------- 初始化zset結(jié)構(gòu)
redis.call("zadd", permitsName, ARGV[2], struct.pack("fI", ARGV[3], ARGV[1]))
-- struct.pack第一個(gè)參數(shù)表示格式字符串，f是浮點(diǎn)數(shù)、I是長(zhǎng)整數(shù)。所以這個(gè)格式字符串表示的是把一個(gè)浮點(diǎn)數(shù)和長(zhǎng)整數(shù)拼起來(lái)的結(jié)構(gòu)體，
-- ARGV[2]就是請(qǐng)求時(shí)間戳，ARGV[1]是請(qǐng)求的令牌數(shù)，統(tǒng)計(jì)會(huì)用到，ARGV[3]是當(dāng)前時(shí)間戳為種子的隨機(jī)數(shù)，具體用處還不知道，知道的網(wǎng)友可以留言
------------------------- 更新Stringt結(jié)構(gòu)，因?yàn)檫@是獲取令牌操作，減掉一個(gè)令牌
-------------------------【本文作者認(rèn)為，這里可以直接初始化string結(jié)構(gòu)，值為rate - 1】
redis.call("decrby", valueName, ARGV[1])
return nil
end
這段lua代碼也并不復(fù)雜，令牌桶的數(shù)量主要是通過(guò)時(shí)間窗口來(lái)控制，判斷上一個(gè)請(qǐng)求是否超過(guò)了令牌生產(chǎn)周期。
留下一個(gè)疑問(wèn)？
-- 移除zset中所有元素【要求是同一個(gè)限流器permitsName，不然就移除不了，尷尬】
redis.call("zrem", permitsName, unpack(expiredValues))
我自己在本地測(cè)試，只要超過(guò)10s，permitsName就不一樣，這就導(dǎo)致了這部分?jǐn)?shù)據(jù)是不能移除的，就產(chǎn)生了冗余數(shù)據(jù)，從前面的截圖也可以看出，是新生成了一個(gè)zset數(shù)據(jù)結(jié)構(gòu)。
相當(dāng)于直接走到了這一步：
------------------------- 更新zset結(jié)構(gòu)
redis.call("zadd", permitsName, ARGV[2], struct.pack("fI", ARGV[3], ARGV[1]))
至于為什么會(huì)產(chǎn)生這樣的結(jié)果，會(huì)的小伙伴可以留言，或者過(guò)段時(shí)間我提個(gè)issue。
以上就是redisson分布式限流RRateLimiter源碼解析的詳細(xì)內(nèi)容，更多關(guān)于redisson分布式限流RRateLimiter的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
Java實(shí)現(xiàn)解析.xlsb文件的示例代碼
這篇文章主要為大家詳細(xì)介紹了Java實(shí)現(xiàn)解析.xlsb文件的相關(guān)方法，文中的示例代碼講解詳細(xì)，具有一定的借鑒價(jià)值，感興趣的可以了解一下2023-01-01
mybatis 映射文件中if標(biāo)簽判斷字符串相等的兩種方式
這篇文章主要介紹了mybatis 映射文件中if標(biāo)簽判斷字符串相等的方式，非常不錯(cuò)，具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2018-06-06
java兩個(gè)線程同時(shí)寫(xiě)一個(gè)文件
這篇文章主要為大家詳細(xì)介紹了java兩個(gè)線程同時(shí)寫(xiě)一個(gè)文件，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2019-08-08
概述java虛擬機(jī)中類的加載器及類加載過(guò)程
這篇文章主要介紹了概述java虛擬機(jī)中類的加載器及類加載過(guò)程,文中有非常詳細(xì)的代碼示例,對(duì)正在學(xué)習(xí)java的小伙伴們有非常好的幫助,需要的朋友可以參考下2021-04-04
MyBatis Plus 實(shí)現(xiàn)多表分頁(yè)查詢功能的示例代碼
這篇文章主要介紹了MyBatis Plus 實(shí)現(xiàn)多表分頁(yè)查詢功能，本文給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2020-08-08
本文主要介紹了Java中幾個(gè)常用的數(shù)據(jù)類型是如何輸入和輸出的，例如：Char型、int型、double型、數(shù)組、字符串等，對(duì)我們學(xué)習(xí)java有一定的幫助，感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)學(xué)習(xí)2021-12-12
這篇文章主要介紹了SpringMVC如何用Post方式重定向，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2022-10-10
java 正則表達(dá)式獲取兩個(gè)字符中間的字符串方法
今天小編就為大家分享一篇java 正則表達(dá)式獲取兩個(gè)字符中間的字符串方法，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2018-07-07
這篇文章主要介紹了Java接口冪等性設(shè)計(jì)原理解析,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-05-05
SpringMVC后端Controller頁(yè)面跳轉(zhuǎn)的三種方式匯總
這篇文章主要介紹了SpringMVC后端Controller頁(yè)面跳轉(zhuǎn)的三種方式匯總,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-10-10