序
原文重要研讨一高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 i妹妹ediately,
* with the value {@code true},
* reducing the number of available permits by one.
*
* <p>If no permit is available then this method will return
* i妹妹ediately 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 i妹妹ediately,
* 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
* i妹妹ediately 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 i妹妹ediately,
* 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 i妹妹ediately, 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 i妹妹ediately,
* 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 i妹妹ediately,
* 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 i妹妹ediately,
* with the value {@code true},
* reducing the number of available permits by one.
*
* <p>If no permit is available then this method will return
* i妹妹ediately 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 i妹妹ediately,
* 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
* i妹妹ediately 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 i妹妹ediately,
* 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 i妹妹ediately, 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 i妹妹ediately,
* 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 i妹妹ediately,
* 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(RedisCo妹妹ands.EVAL_NULL_BOOLEAN, permits));
}
@Override
public RFuture<Boolean> tryAcquireAsync(long permits) {
return tryAcquireAsync(RedisCo妹妹ands.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，完成了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 co妹妹andExecutor.evalWriteNoRetryAsync(getRawName(), LongCodec.INSTANCE, RedisCo妹妹ands.EVAL_BOOLEAN,
"redis.call('hsetnx', KEYS[1], 'rate', ARGV[1]);"
+ "redis.call('hsetnx', KEYS[1], 'interval', ARGV[二]);"
+ "return redis.call('hsetnx', KEYS[1], 'type', ARGV[3]);",
Collections.singletonList(getRawName()), rate, unit.toMillis(rateInterval), type.ordinal());
}
trySetRate委托给了trySetRateAsync，那面重要是利用hsetnx来装置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 co妹妹andExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCo妹妹ands.EVAL_BOOLEAN,
"local valueName = KEYS[两];"
+ "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[两]);"
+ "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(RedisCo妹妹ands.EVAL_NULL_BOOLEAN, permits));
}
private <T> RFuture<T> tryAcquireAsync(RedisCo妹妹and<T> co妹妹and, Long value) {
byte[] random = getServiceManager().generateIdArray();
return co妹妹andExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, co妹妹and,
"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[两];"
+ "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[二]) - 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[二]) - 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[二]) - tonumber(firstValue[二]));"
+ "else "
+ "redis.call('zadd', permitsName, ARGV[两], 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[两], 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剧本来执止，起首经由过程hget猎取rate、interval、type的值，而后按照type来确定valueName、permitsName，假如type为0则valueName是getValueName()，permitsName是getPermitsName()，假如type=1则valueName是getClientValueName()，permitsName是getClientPermitsName()；以后猎取valueName的值，若为false则间接用部署rate、permits，并递加valueName；若为true则猎取expiredValues算计released值，再算计没currentValue，若不足扣则计较返归值，若够扣则经由过程zadd加添当前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(RedisCo妹妹ands.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);
co妹妹andExecutor.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);
co妹妹andExecutor.transfer(r, f);
}, delay, TimeUnit.MILLISECONDS);
}
return f;
}).toCompletableFuture();
}
acquire也是复用了tryAcquireAsync办法，只猎取没有到时会按照返归的delay入止从新调度，若timeoutInMillis没有为-1则会按照超时光阴入止计较以及从新调度
availablePermits
public long availablePermits() {
return get(availablePermitsAsync());
}
public RFuture<Long> availablePermitsAsync() {
return co妹妹andExecutor.evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCo妹妹ands.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[两];"
+ "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，它执止lua剧本，先经由过程hget猎取rate、interval、type的值，而后依照type来确定valueName、permitsName，若何type为0则valueName是getValueName()，permitsName是getPermitsName()，假如type=1则valueName是getClientValueName()，permitsName是getClientPermitsName()；以后猎取valueName对于应的值currentValue，若值为false则从新陈设rate，不然经由过程expiredValues从新计较released，若released年夜于0则更新到currentValue，末了返归currentValue
年夜结
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，存储了当前的permits；一个是key为permitsName，范例是sorted set，其score为System.currentTimeMillis()，value经由过程struct.pack了随机数少度、随机数、这次permit的value
- trySetRate委托给了trySetRateAsync，那面首要是利用hsetnx来陈设rate、interval、type三个值；setRate委托给了setRateAsync，那面应用hset来写进rate、interval、type三个值，假如具有则笼盖；此外那面增除了了valueName、permitsName那二个key
- tryAcquire委托给了tryAcquireAsync，它经由过程一个lua剧本来执止，起首经由过程hget猎取rate、interval、type的值，以后猎取valueName的值，若为false则间接用配置rate、permits，并递加valueName；若为true则猎取expiredValues算计released值，再算计没currentValue，若不敷扣则计较返归值(汇报挪用圆否以延时多永劫间再重试)，若够扣则经由过程zadd加添当前permit(
System.currentTimeMillis())，而后递加valueName - acquire是也是复用了tryAcquireAsync办法，只猎取没有到时会按照返归的delay入止从新调度，若timeoutInMillis没有为-1则会按照超时光阴入止计较以及从新调度
到此那篇闭于redisson外RRateLimiter漫衍式限流器的利用的文章便先容到那了,更多相闭redisson RRateLimiter形式请搜刮剧本之野之前的文章或者连续涉猎上面的相闭文章心愿大家2之后多多撑持剧本之野！
发表评论 取消回复