redisson分布式限流RRateLimiter源碼解析
更新時間：2022年11月01日 16:15:56 作者：你行你上啊
這篇文章主要為大家介紹了redisson分布式限流RRateLimiter源碼解析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪
分布式限流-單位時間多實例多線程訪問次數(shù)限制
接前面聊一聊redisson及優(yōu)雅實現(xiàn) 和 說一說spring boot優(yōu)雅集成redisson，簡單以源碼的方式給大家介紹了redisson的：可重入性、阻塞、續(xù)約、紅鎖、聯(lián)鎖、加鎖解鎖流程和集成spring boot注意點和優(yōu)雅實現(xiàn)方式。
接下來在講一講平時用的比較多的限流模塊--RRateLimiter
1.簡單使用
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 單實例執(zhí)行，OVERALL 全實例執(zhí)行
// 最大流速 = 每10秒鐘產(chǎn)生3個令牌
rateLimiter.trySetRate(RateType.OVERALL, 3, 10, RateIntervalUnit.SECONDS);
return rateLimiter;
}
實際結果：
[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. 實現(xiàn)限流redisson使用了哪些redis數(shù)據(jù)結構
- Hash結構 -- 限流器結構：
參數(shù)rate代表速率
參數(shù)interval代表多少時間內(nèi)產(chǎn)生的令牌
參數(shù)type代表單機還是集群
- ZSET結構 -- 記錄獲取令牌的時間戳，用于時間對比。
1667025166312 --> 2022-10-29 14:32:46
1667025166262 --> 2022-10-29 14:32:46
1667025166215 --> 2022-10-29 14:32:46
- String結構 --記錄的是當前令牌桶中的令牌數(shù)【很明顯被我用完了現(xiàn)在是0】
3. 超過10s,我再次獲取一個令牌，數(shù)據(jù)結構發(fā)生的變化
- ZSET結構。-- 新生成一個ZSET結構，存放獲取令牌的時間戳
- String 結構 --當前令牌桶還有2個令牌
4. 源碼淺析
RRateLimiter rateLimiter = redisson.getRateLimiter("myRateLimiter3");
// 初始化
// 最大流速 = 每10秒鐘產(chǎn)生3個令牌
rateLimiter.trySetRate(RateType.PER_CLIENT, 3, 10, RateIntervalUnit.SECONDS);
初始化定義沒有什么好講的，就是創(chuàng)建HASH結構
主要還是講講： 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代碼，下面我詳細過一下
作者目前用的3.16.3版本，剛好遇見redisson的bug，見3197，請大家用最新版本，以下為修復后解析。
-- 獲取hash結構的速率
local rate = redis.call("hget", KEYS[1], "rate")
-- 獲取hash結構的時間區(qū)間(ms)
local interval = redis.call("hget", KEYS[1], "interval")
-- 獲取hash結構的時間類型
local type = redis.call("hget", KEYS[1], "type")
-- 判斷是否初始化限流結構
assert(rate ~= false and interval ~= false and type ~= false, "RateLimiter is not initialized")
-- {name}:value string結構，這個key記錄的是當前令牌桶中的令牌數(shù)
local valueName = KEYS[2]
-- {name}:permits zset結構，記錄了請求的令牌數(shù)，score則為請求的時間戳
local permitsName = KEYS[4]
-- 單機限流才會用到，集群模式不用關注
if type == "1" then
valueName = KEYS[3]
permitsName = KEYS[5]
end
-- 生產(chǎn)速率rate必須比請求的令牌數(shù)大
assert(tonumber(rate) >= tonumber(ARGV[1]), "Requested permits amount could not exceed defined rate")
-- 初始化RateLimiter并不會初始化stirng結構，因此第一次獲取這里currentValue是null
local currentValue = redis.call("get", valueName)
if currentValue ~= false then
-- 第二次獲取令牌執(zhí)行
-------------------------- 獲取zset結構：統(tǒng)計之前的請求令牌數(shù)
-- 范圍是0 ~ (第二次請求時間戳 - 令牌生產(chǎn)的時間)
local expiredValues = redis.call("zrangebyscore", permitsName, 0, tonumber(ARGV[2]) - interval)
local released = 0
-- lua迭代器，遍歷expiredValues，如果有值，那么released等于之前所有請求的令牌數(shù)之和，表示應該釋放多少令牌
for i, v in ipairs(expiredValues) do
-- 獲取請求數(shù)permits
local random, permits = struct.unpack("fI", v)
released = released + permits
end
-- 之前的請求令牌數(shù) > 0, 例如10s產(chǎn)生3個令牌，現(xiàn)在超過10s了，重置周期并計算剩余令牌數(shù)
if released > 0 then
-- 移除zset中所有元素【要求是同一個限流器permitsName，不然就移除不了，尷尬】
redis.call("zrem", permitsName, unpack(expiredValues))
currentValue = tonumber(currentValue) + released
------------------------- 更新string結構：=剩下令牌數(shù)+釋放令牌數(shù)
redis.call("set", valueName, currentValue)
end
-- 如果當前令牌數(shù) ＜ 請求的令牌數(shù)
if tonumber(currentValue) < tonumber(ARGV[1]) then
-- 從zset中找到距離當前時間最近的那個請求，也就是上一次放進去的請求信息
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])
-- 返回 上一次請求的時間戳 - (當前時間戳 - 令牌生成的時間間隔) 這個值表示還需要多久才能生產(chǎn)出足夠的令牌
return tonumber(nearest[2]) - (tonumber(ARGV[2]) - interval)
else
-- 如果當前令牌數(shù) ≥ 請求的令牌數(shù)，表示令牌夠多，更新zset
------------------------- 更新zset結構
redis.call("zadd", permitsName, ARGV[2], struct.pack("fI", ARGV[3], ARGV[1]))
------------------------- 更新Stringt結構，減少一個剩下的令牌數(shù)
redis.call("decrby", valueName, ARGV[1])
return nil
end
else
--------汀雨筆記----------------- 初始化Stringt結構，當前限流器的令牌數(shù)
redis.call("set", valueName, rate)
--------汀雨筆記----------------- 初始化zset結構
redis.call("zadd", permitsName, ARGV[2], struct.pack("fI", ARGV[3], ARGV[1]))
-- struct.pack第一個參數(shù)表示格式字符串，f是浮點數(shù)、I是長整數(shù)。所以這個格式字符串表示的是把一個浮點數(shù)和長整數(shù)拼起來的結構體，
-- ARGV[2]就是請求時間戳，ARGV[1]是請求的令牌數(shù)，統(tǒng)計會用到，ARGV[3]是當前時間戳為種子的隨機數(shù)，具體用處還不知道，知道的網(wǎng)友可以留言
------------------------- 更新Stringt結構，因為這是獲取令牌操作，減掉一個令牌
-------------------------【本文作者認為，這里可以直接初始化string結構，值為rate - 1】
redis.call("decrby", valueName, ARGV[1])
return nil
end
這段lua代碼也并不復雜，令牌桶的數(shù)量主要是通過時間窗口來控制，判斷上一個請求是否超過了令牌生產(chǎn)周期。
留下一個疑問？
-- 移除zset中所有元素【要求是同一個限流器permitsName，不然就移除不了，尷尬】
redis.call("zrem", permitsName, unpack(expiredValues))
我自己在本地測試，只要超過10s，permitsName就不一樣，這就導致了這部分數(shù)據(jù)是不能移除的，就產(chǎn)生了冗余數(shù)據(jù)，從前面的截圖也可以看出，是新生成了一個zset數(shù)據(jù)結構。
相當于直接走到了這一步：
------------------------- 更新zset結構
redis.call("zadd", permitsName, ARGV[2], struct.pack("fI", ARGV[3], ARGV[1]))
至于為什么會產(chǎn)生這樣的結果，會的小伙伴可以留言，或者過段時間我提個issue。
以上就是redisson分布式限流RRateLimiter源碼解析的詳細內(nèi)容，更多關于redisson分布式限流RRateLimiter的資料請關注腳本之家其它相關文章！
相關文章
springboot3整合SpringSecurity實現(xiàn)登錄校驗與權限認證
本文主要介紹了springboot3整合SpringSecurity實現(xiàn)登錄校驗與權限認證,文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友們下面隨著小編來一起學習學習吧2025-04-04
SSH框架網(wǎng)上商城項目第24戰(zhàn)之Struts2中處理多個Model請求的方法
這篇文章主要為大家詳細介紹了SSH框架網(wǎng)上商城項目第24戰(zhàn)之Struts2中處理多個Model請求的方法，感興趣的小伙伴們可以參考一下2016-06-06
SpringBoot?jackson提供對LocalDate的支持方式
這篇文章主要介紹了SpringBoot?jackson提供對LocalDate的支持方式，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-01-01
Spring Boot打開URL出現(xiàn)signin問題的解決
這篇文章主要介紹了Spring Boot打開URL出現(xiàn)signin問題的解決，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-12-12
在Java中,我們都知道直接交換Map的key和value是不被允許的,因為Map的接口設計是基于key-value對的,其中key是唯一的,并且是不可變的,所以本文給大家介紹了Java交換map的key和value值的步驟和代碼示例,需要的朋友可以參考下2024-09-09
Java中Integer.valueOf,parsetInt() String.valueOf的區(qū)別和結果代碼解析
本文通過代碼給大家講解了JAVA中Integer.valueOf， parsetInt() String.valueOf的區(qū)別和結果,需要的朋友可以參考下2018-05-05
Java實戰(zhàn)之晚會抽獎系統(tǒng)的實現(xiàn)
這篇文章主要介紹了如何利用Java語言編寫一個晚會抽獎系統(tǒng)，文中采用到的技術有Jdbc、Servlert、JavaScript、JQuery、Ajax等，感興趣的可以學習一下2022-03-03
這篇文章主要介紹了Spring 項目常用pom文件的依賴，文中給大家提到了Spring boot starter pom的依賴關系說明，需要的朋友參考下吧2018-03-03
解決SpringMvc中普通類注入Service為null的問題
這篇文章主要介紹了解決SpringMvc中普通類注入Service為null的問題，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-07-07
這篇文章主要介紹了Java多線程中的ThreadPoolExecutor解讀,線程池中的核心線程數(shù),當提交一個任務時,線程池創(chuàng)建一個新線程執(zhí)行任務,直到當前線程數(shù)等于corePoolSize；如果當前線程數(shù)為corePoolSize,繼續(xù)提交的任務被保存到阻塞隊列中,等待被執(zhí)行,需要的朋友可以參考下2023-09-09