redisson中RRateLimiter分布式限流器的使用
序
本文主要研究一下redisson的RRateLimiter
RRateLimiter
redisson/src/main/java/org/redisson/api/RRateLimiter.java
public interface RRateLimiter extends RRateLimiterAsync, RExpirable {
/**
* Initializes RateLimiter's state and stores config to Redis server.
*
* @param mode - rate mode
* @param rate - rate
* @param rateInterval - rate time interval
* @param rateIntervalUnit - rate time interval unit
* @return {@code true} if rate was set and {@code false}
* otherwise
*/
boolean trySetRate(RateType mode, long rate, long rateInterval, RateIntervalUnit rateIntervalUnit);
/**
* Updates RateLimiter's state and stores config to Redis server.
*
* @param mode - rate mode
* @param rate - rate
* @param rateInterval - rate time interval
* @param rateIntervalUnit - rate time interval unit
*/
void setRate(RateType mode, long rate, long rateInterval, RateIntervalUnit rateIntervalUnit);
/**
* Acquires a permit only if one is available at the
* time of invocation.
*
* <p>Acquires a permit, if one is available and returns immediately,
* with the value {@code true},
* reducing the number of available permits by one.
*
* <p>If no permit is available then this method will return
* immediately with the value {@code false}.
*
* @return {@code true} if a permit was acquired and {@code false}
* otherwise
*/
boolean tryAcquire();
/**
* Acquires the given number of <code>permits</code> only if all are available at the
* time of invocation.
*
* <p>Acquires a permits, if all are available and returns immediately,
* with the value {@code true},
* reducing the number of available permits by given number of permits.
*
* <p>If no permits are available then this method will return
* immediately with the value {@code false}.
*
* @param permits the number of permits to acquire
* @return {@code true} if a permit was acquired and {@code false}
* otherwise
*/
boolean tryAcquire(long permits);
/**
* Acquires a permit from this RateLimiter, blocking until one is available.
*
* <p>Acquires a permit, if one is available and returns immediately,
* reducing the number of available permits by one.
*
*/
void acquire();
/**
* Acquires a specified <code>permits</code> from this RateLimiter,
* blocking until one is available.
*
* <p>Acquires the given number of permits, if they are available
* and returns immediately, reducing the number of available permits
* by the given amount.
*
* @param permits the number of permits to acquire
*/
void acquire(long permits);
/**
* Acquires a permit from this RateLimiter, if one becomes available
* within the given waiting time.
*
* <p>Acquires a permit, if one is available and returns immediately,
* with the value {@code true},
* reducing the number of available permits by one.
*
* <p>If no permit is available then the current thread becomes
* disabled for thread scheduling purposes and lies dormant until
* specified waiting time elapses.
*
* <p>If a permit is acquired then the value {@code true} is returned.
*
* <p>If the specified waiting time elapses then the value {@code false}
* is returned. If the time is less than or equal to zero, the method
* will not wait at all.
*
* @param timeout the maximum time to wait for a permit
* @param unit the time unit of the {@code timeout} argument
* @return {@code true} if a permit was acquired and {@code false}
* if the waiting time elapsed before a permit was acquired
*/
boolean tryAcquire(long timeout, TimeUnit unit);
/**
* Acquires the given number of <code>permits</code> only if all are available
* within the given waiting time.
*
* <p>Acquires the given number of permits, if all are available and returns immediately,
* with the value {@code true}, reducing the number of available permits by one.
*
* <p>If no permit is available then the current thread becomes
* disabled for thread scheduling purposes and lies dormant until
* the specified waiting time elapses.
*
* <p>If a permits is acquired then the value {@code true} is returned.
*
* <p>If the specified waiting time elapses then the value {@code false}
* is returned. If the time is less than or equal to zero, the method
* will not wait at all.
*
* @param permits amount
* @param timeout the maximum time to wait for a permit
* @param unit the time unit of the {@code timeout} argument
* @return {@code true} if a permit was acquired and {@code false}
* if the waiting time elapsed before a permit was acquired
*/
boolean tryAcquire(long permits, long timeout, TimeUnit unit);
/**
* Returns current configuration of this RateLimiter object.
*
* @return config object
*/
RateLimiterConfig getConfig();
/**
* Returns amount of available permits.
*
* @return number of permits
*/
long availablePermits();
}
RRateLimiter繼承了RRateLimiterAsync、RExpirable接口,它主要定義了trySetRate、setRate、tryAcquire、acquire、getConfig、availablePermits方法
RRateLimiterAsync
redisson/src/main/java/org/redisson/api/RRateLimiterAsync.java
public interface RRateLimiterAsync extends RExpirableAsync {
/**
* Initializes RateLimiter's state and stores config to Redis server.
*
* @param mode - rate mode
* @param rate - rate
* @param rateInterval - rate time interval
* @param rateIntervalUnit - rate time interval unit
* @return {@code true} if rate was set and {@code false}
* otherwise
*/
RFuture<Boolean> trySetRateAsync(RateType mode, long rate, long rateInterval, RateIntervalUnit rateIntervalUnit);
/**
* Acquires a permit only if one is available at the
* time of invocation.
*
* <p>Acquires a permit, if one is available and returns immediately,
* with the value {@code true},
* reducing the number of available permits by one.
*
* <p>If no permit is available then this method will return
* immediately with the value {@code false}.
*
* @return {@code true} if a permit was acquired and {@code false}
* otherwise
*/
RFuture<Boolean> tryAcquireAsync();
/**
* Acquires the given number of <code>permits</code> only if all are available at the
* time of invocation.
*
* <p>Acquires a permits, if all are available and returns immediately,
* with the value {@code true},
* reducing the number of available permits by given number of permits.
*
* <p>If no permits are available then this method will return
* immediately with the value {@code false}.
*
* @param permits the number of permits to acquire
* @return {@code true} if a permit was acquired and {@code false}
* otherwise
*/
RFuture<Boolean> tryAcquireAsync(long permits);
/**
* Acquires a permit from this RateLimiter, blocking until one is available.
*
* <p>Acquires a permit, if one is available and returns immediately,
* reducing the number of available permits by one.
*
* @return void
*/
RFuture<Void> acquireAsync();
/**
* Acquires a specified <code>permits</code> from this RateLimiter,
* blocking until one is available.
*
* <p>Acquires the given number of permits, if they are available
* and returns immediately, reducing the number of available permits
* by the given amount.
*
* @param permits the number of permits to acquire
* @return void
*/
RFuture<Void> acquireAsync(long permits);
/**
* Acquires a permit from this RateLimiter, if one becomes available
* within the given waiting time.
*
* <p>Acquires a permit, if one is available and returns immediately,
* with the value {@code true},
* reducing the number of available permits by one.
*
* <p>If no permit is available then the current thread becomes
* disabled for thread scheduling purposes and lies dormant until
* specified waiting time elapses.
*
* <p>If a permit is acquired then the value {@code true} is returned.
*
* <p>If the specified waiting time elapses then the value {@code false}
* is returned. If the time is less than or equal to zero, the method
* will not wait at all.
*
* @param timeout the maximum time to wait for a permit
* @param unit the time unit of the {@code timeout} argument
* @return {@code true} if a permit was acquired and {@code false}
* if the waiting time elapsed before a permit was acquired
*/
RFuture<Boolean> tryAcquireAsync(long timeout, TimeUnit unit);
/**
* Acquires the given number of <code>permits</code> only if all are available
* within the given waiting time.
*
* <p>Acquires the given number of permits, if all are available and returns immediately,
* with the value {@code true}, reducing the number of available permits by one.
*
* <p>If no permit is available then the current thread becomes
* disabled for thread scheduling purposes and lies dormant until
* the specified waiting time elapses.
*
* <p>If a permits is acquired then the value {@code true} is returned.
*
* <p>If the specified waiting time elapses then the value {@code false}
* is returned. If the time is less than or equal to zero, the method
* will not wait at all.
*
* @param permits amount
* @param timeout the maximum time to wait for a permit
* @param unit the time unit of the {@code timeout} argument
* @return {@code true} if a permit was acquired and {@code false}
* if the waiting time elapsed before a permit was acquired
*/
RFuture<Boolean> tryAcquireAsync(long permits, long timeout, TimeUnit unit);
/**
* Updates RateLimiter's state and stores config to Redis server.
*
*
* @param mode - rate mode
* @param rate - rate
* @param rateInterval - rate time interval
* @param rateIntervalUnit - rate time interval unit
* @return {@code true} if rate was set and {@code false}
* otherwise
*/
RFuture<Void> setRateAsync(RateType mode, long rate, long rateInterval, RateIntervalUnit rateIntervalUnit);
/**
* Returns current configuration of this RateLimiter object.
*
* @return config object
*/
RFuture<RateLimiterConfig> getConfigAsync();
/**
* Returns amount of available permits.
*
* @return number of permits
*/
RFuture<Long> availablePermitsAsync();
}
RRateLimiterAsync繼承了RExpirableAsync,它是async版本的RRateLimiter,它主要定義了trySetRateAsync、setRateAsync、tryAcquireAsync、acquireAsync、getConfigAsync、availablePermitsAsync方法
RedissonRateLimiter
redisson/src/main/java/org/redisson/RedissonRateLimiter.java
public class RedissonRateLimiter extends RedissonExpirable implements RRateLimiter {
//......
@Override
public boolean tryAcquire() {
return tryAcquire(1);
}
@Override
public RFuture<Boolean> tryAcquireAsync() {
return tryAcquireAsync(1L);
}
@Override
public boolean tryAcquire(long permits) {
return get(tryAcquireAsync(RedisCommands.EVAL_NULL_BOOLEAN, permits));
}
@Override
public RFuture<Boolean> tryAcquireAsync(long permits) {
return tryAcquireAsync(RedisCommands.EVAL_NULL_BOOLEAN, permits);
}
@Override
public void acquire() {
get(acquireAsync());
}
@Override
public RFuture<Void> acquireAsync() {
return acquireAsync(1);
}
@Override
public void acquire(long permits) {
get(acquireAsync(permits));
}
@Override
public RFuture<Void> acquireAsync(long permits) {
CompletionStage<Void> f = tryAcquireAsync(permits, -1, null).thenApply(res -> null);
return new CompletableFutureWrapper<>(f);
}
@Override
public boolean tryAcquire(long timeout, TimeUnit unit) {
return get(tryAcquireAsync(timeout, unit));
}
@Override
public RFuture<Boolean> tryAcquireAsync(long timeout, TimeUnit unit) {
return tryAcquireAsync(1, timeout, unit);
}
@Override
public boolean tryAcquire(long permits, long timeout, TimeUnit unit) {
return get(tryAcquireAsync(permits, timeout, unit));
}
}
RedissonRateLimiter繼承了RedissonExpirable,實現(xiàn)了RRateLimiter接口
trySetRate
public boolean trySetRate(RateType type, long rate, long rateInterval, RateIntervalUnit unit) {
return get(trySetRateAsync(type, rate, rateInterval, unit));
}
public RFuture<Boolean> trySetRateAsync(RateType type, long rate, long rateInterval, RateIntervalUnit unit) {
return commandExecutor.evalWriteNoRetryAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"redis.call('hsetnx', KEYS[1], 'rate', ARGV[1]);"
+ "redis.call('hsetnx', KEYS[1], 'interval', ARGV[2]);"
+ "return redis.call('hsetnx', KEYS[1], 'type', ARGV[3]);",
Collections.singletonList(getRawName()), rate, unit.toMillis(rateInterval), type.ordinal());
}
trySetRate委托給了trySetRateAsync,這里主要是使用hsetnx來設(shè)置rate、interval、type三個值
setRate
public void setRate(RateType type, long rate, long rateInterval, RateIntervalUnit unit) {
get(setRateAsync(type, rate, rateInterval, unit));
}
public RFuture<Void> setRateAsync(RateType type, long rate, long rateInterval, RateIntervalUnit unit) {
return commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
"local valueName = KEYS[2];"
+ "local permitsName = KEYS[4];"
+ "if ARGV[3] == '1' then "
+ " valueName = KEYS[3];"
+ " permitsName = KEYS[5];"
+ "end "
+"redis.call('hset', KEYS[1], 'rate', ARGV[1]);"
+ "redis.call('hset', KEYS[1], 'interval', ARGV[2]);"
+ "redis.call('hset', KEYS[1], 'type', ARGV[3]);"
+ "redis.call('del', valueName, permitsName);",
Arrays.asList(getRawName(), getValueName(), getClientValueName(), getPermitsName(), getClientPermitsName()), rate, unit.toMillis(rateInterval), type.ordinal());
}
setRate委托給了setRateAsync,這里使用hset來寫入rate、interval、type三個值,如果存在則覆蓋;另外這里刪除了valueName、permitsName這兩個key
tryAcquire
public boolean tryAcquire(long permits) {
return get(tryAcquireAsync(RedisCommands.EVAL_NULL_BOOLEAN, permits));
}
private <T> RFuture<T> tryAcquireAsync(RedisCommand<T> command, Long value) {
byte[] random = getServiceManager().generateIdArray();
return commandExecutor.evalWriteAsync(getRawName(), 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;"
+ "assert(tonumber(rate) >= tonumber(ARGV[1]), 'Requested permits amount could not exceed defined rate'); "
+ "local currentValue = redis.call('get', valueName); "
+ "local res;"
+ "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('Bc0I', v);"
+ "released = released + permits;"
+ "end; "
+ "if released > 0 then "
+ "redis.call('zremrangebyscore', permitsName, 0, tonumber(ARGV[2]) - interval); "
+ "if tonumber(currentValue) + released > tonumber(rate) then "
+ "currentValue = tonumber(rate) - redis.call('zcard', permitsName); "
+ "else "
+ "currentValue = tonumber(currentValue) + released; "
+ "end; "
+ "redis.call('set', valueName, currentValue);"
+ "end;"
+ "if tonumber(currentValue) < tonumber(ARGV[1]) then "
+ "local firstValue = redis.call('zrange', permitsName, 0, 0, 'withscores'); "
+ "res = 3 + interval - (tonumber(ARGV[2]) - tonumber(firstValue[2]));"
+ "else "
+ "redis.call('zadd', permitsName, ARGV[2], struct.pack('Bc0I', string.len(ARGV[3]), ARGV[3], ARGV[1])); "
+ "redis.call('decrby', valueName, ARGV[1]); "
+ "res = nil; "
+ "end; "
+ "else "
+ "redis.call('set', valueName, rate); "
+ "redis.call('zadd', permitsName, ARGV[2], struct.pack('Bc0I', string.len(ARGV[3]), ARGV[3], ARGV[1])); "
+ "redis.call('decrby', valueName, ARGV[1]); "
+ "res = nil; "
+ "end;"
+ "local ttl = redis.call('pttl', KEYS[1]); "
+ "if ttl > 0 then "
+ "redis.call('pexpire', valueName, ttl); "
+ "redis.call('pexpire', permitsName, ttl); "
+ "end; "
+ "return res;",
Arrays.asList(getRawName(), getValueName(), getClientValueName(), getPermitsName(), getClientPermitsName()),
value, System.currentTimeMillis(), random);
}
tryAcquire委托給了tryAcquireAsync,它通過一個lua腳本來執(zhí)行,首先通過hget獲取rate、interval、type的值,然后根據(jù)type來確定valueName、permitsName,如果type為0則valueName是getValueName(),permitsName是getPermitsName(),如果type=1則valueName是getClientValueName(),permitsName是getClientPermitsName();之后獲取valueName的值,若為false則直接用設(shè)置rate、permits,并遞減valueName;若為true則獲取expiredValues計算released值,再計算出currentValue,若不夠扣則計算返回值,若夠扣則通過zadd添加當(dāng)前permit(
System.currentTimeMillis()),然后遞減valueName
acquire
public void acquire() {
get(acquireAsync());
}
public RFuture<Void> acquireAsync() {
return acquireAsync(1);
}
public RFuture<Void> acquireAsync(long permits) {
CompletionStage<Void> f = tryAcquireAsync(permits, -1, null).thenApply(res -> null);
return new CompletableFutureWrapper<>(f);
}
public RFuture<Boolean> tryAcquireAsync(long permits, long timeout, TimeUnit unit) {
long timeoutInMillis = -1;
if (timeout >= 0) {
timeoutInMillis = unit.toMillis(timeout);
}
CompletableFuture<Boolean> f = tryAcquireAsync(permits, timeoutInMillis);
return new CompletableFutureWrapper<>(f);
}
private CompletableFuture<Boolean> tryAcquireAsync(long permits, long timeoutInMillis) {
long s = System.currentTimeMillis();
RFuture<Long> future = tryAcquireAsync(RedisCommands.EVAL_LONG, permits);
return future.thenCompose(delay -> {
if (delay == null) {
return CompletableFuture.completedFuture(true);
}
if (timeoutInMillis == -1) {
CompletableFuture<Boolean> f = new CompletableFuture<>();
getServiceManager().getGroup().schedule(() -> {
CompletableFuture<Boolean> r = tryAcquireAsync(permits, timeoutInMillis);
commandExecutor.transfer(r, f);
}, delay, TimeUnit.MILLISECONDS);
return f;
}
long el = System.currentTimeMillis() - s;
long remains = timeoutInMillis - el;
if (remains <= 0) {
return CompletableFuture.completedFuture(false);
}
CompletableFuture<Boolean> f = new CompletableFuture<>();
if (remains < delay) {
getServiceManager().getGroup().schedule(() -> {
f.complete(false);
}, remains, TimeUnit.MILLISECONDS);
} else {
long start = System.currentTimeMillis();
getServiceManager().getGroup().schedule(() -> {
long elapsed = System.currentTimeMillis() - start;
if (remains <= elapsed) {
f.complete(false);
return;
}
CompletableFuture<Boolean> r = tryAcquireAsync(permits, remains - elapsed);
commandExecutor.transfer(r, f);
}, delay, TimeUnit.MILLISECONDS);
}
return f;
}).toCompletableFuture();
}
acquire也是復(fù)用了tryAcquireAsync方法,只獲取不到時會根據(jù)返回的delay進(jìn)行重新調(diào)度,若timeoutInMillis不為-1則會根據(jù)超時時間進(jìn)行計算和重新調(diào)度
availablePermits
public long availablePermits() {
return get(availablePermitsAsync());
}
public RFuture<Long> availablePermitsAsync() {
return commandExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_LONG,
"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 "
+ "redis.call('set', valueName, rate); "
+ "return rate; "
+ "else "
+ "local expiredValues = redis.call('zrangebyscore', permitsName, 0, tonumber(ARGV[1]) - interval); "
+ "local released = 0; "
+ "for i, v in ipairs(expiredValues) do "
+ "local random, permits = struct.unpack('Bc0I', v);"
+ "released = released + permits;"
+ "end; "
+ "if released > 0 then "
+ "redis.call('zremrangebyscore', permitsName, 0, tonumber(ARGV[1]) - interval); "
+ "currentValue = tonumber(currentValue) + released; "
+ "redis.call('set', valueName, currentValue);"
+ "end;"
+ "return currentValue; "
+ "end;",
Arrays.asList(getRawName(), getValueName(), getClientValueName(), getPermitsName(), getClientPermitsName()),
System.currentTimeMillis());
}
availablePermits委托給了availablePermitsAsync,它執(zhí)行l(wèi)ua腳本,先通過hget獲取rate、interval、type的值,然后根據(jù)type來確定valueName、permitsName,如果type為0則valueName是getValueName(),permitsName是getPermitsName(),如果type=1則valueName是getClientValueName(),permitsName是getClientPermitsName();之后獲取valueName對應(yīng)的值currentValue,若值為false則重新設(shè)置rate,否則通過expiredValues重新計算released,若released大于0則更新到currentValue,最后返回currentValue
小結(jié)
redisson的RRateLimiter提供了trySetRate、setRate、tryAcquire、acquire、getConfig、availablePermits方法
- 其RateType有OVERALL(
值為0)、PER_CLIENT(值為1)兩個類型,如果type為0則valueName是getValueName(),permitsName是getPermitsName(),如果type=1則valueName是getClientValueName(),permitsName是getClientPermitsName() - 它主要定義了幾個key,一個是getRawName,類型為hash,其key有rate、interval、type;一個是key為valueName,存儲了當(dāng)前的permits;一個是key為permitsName,類型是sorted set,其score為System.currentTimeMillis(),value通過struct.pack了隨機(jī)數(shù)長度、隨機(jī)數(shù)、此次permit的value
- trySetRate委托給了trySetRateAsync,這里主要是使用hsetnx來設(shè)置rate、interval、type三個值;setRate委托給了setRateAsync,這里使用hset來寫入rate、interval、type三個值,如果存在則覆蓋;另外這里刪除了valueName、permitsName這兩個key
- tryAcquire委托給了tryAcquireAsync,它通過一個lua腳本來執(zhí)行,首先通過hget獲取rate、interval、type的值,之后獲取valueName的值,若為false則直接用設(shè)置rate、permits,并遞減valueName;若為true則獲取expiredValues計算released值,再計算出currentValue,若不夠扣則計算返回值(告訴調(diào)用方可以延時多長時間再重試),若夠扣則通過zadd添加當(dāng)前permit(
System.currentTimeMillis()),然后遞減valueName - acquire是也是復(fù)用了tryAcquireAsync方法,只獲取不到時會根據(jù)返回的delay進(jìn)行重新調(diào)度,若timeoutInMillis不為-1則會根據(jù)超時時間進(jìn)行計算和重新調(diào)度
到此這篇關(guān)于redisson中RRateLimiter分布式限流器的使用的文章就介紹到這了,更多相關(guān)redisson RRateLimiter內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Spark刪除redis千萬級別set集合數(shù)據(jù)實現(xiàn)分析
這篇文章主要為大家介紹了Spark刪除redis千萬級別set集合數(shù)據(jù)實現(xiàn)過程分析,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2023-06-06
redis實現(xiàn)sentinel哨兵架構(gòu)的方法
哨兵是一個分布式系統(tǒng),可以在一個架構(gòu)中運(yùn)行多個哨兵(sentinel) 進(jìn)程,這些進(jìn)程使用流言協(xié)議(gossip protocols)來接收關(guān)于Master主服務(wù)器是否下線的信息,這篇文章主要介紹了redis實現(xiàn)sentinel哨兵架構(gòu),需要的朋友可以參考下2022-11-11

