Of Stack

Example of redis redisson Current Limiter of RRateLimiter

  • 2021-10-16 01:42:34
  • OfStack

redis redisson Current Limiter Example

Function: Limit the number of accesses to data in a period of time

Related interface

RRateLimiter 


public interface RRateLimiter extends RRateLimiterAsync, RObject { 
    boolean trySetRate(RateType var1, long var2, long var4, RateIntervalUnit var6);
                              // Set the access rate, var2 For the number of accesses, var4 Is unit time, var6 Is the unit of time  
    void acquire();           // Access data 
    void acquire(long var1);  // Account for var1 Calculated value of speed of  
    boolean tryAcquire();                                    // Attempt to access data 
    boolean tryAcquire(long var1);                           // Attempt to access data, accounting for var1 Calculated value of speed of 
    boolean tryAcquire(long var1, TimeUnit var3);            // Try to access data and set the waiting time var3
    boolean tryAcquire(long var1, long var3, TimeUnit var5); // Attempt to access data, accounting for the calculated value of data var1 Set the waiting time var3 
    RateLimiterConfig getConfig();
}

RateType Speed type 


public enum RateType {
    OVERALL,             // Total current limit for all clients 
    PER_CLIENT;          // Each client calculates traffic separately 
 
    private RateType() {
    }
}

RateInternalUnit Velocity unit 


public enum RateIntervalUnit {
    MILLISECONDS {
        public long toMillis(long value) {
            return value;
        }
    },
    SECONDS {
        public long toMillis(long value) {
            return TimeUnit.SECONDS.toMillis(value);
        }
    },
    MINUTES {
        public long toMillis(long value) {
            return TimeUnit.MINUTES.toMillis(value);
        }
    },
    HOURS {
        public long toMillis(long value) {
            return TimeUnit.HOURS.toMillis(value);
        }
    },
    DAYS {
        public long toMillis(long value) {
            return TimeUnit.DAYS.toMillis(value);
        }
    };
 
    private RateIntervalUnit() {
    } 
    public abstract long toMillis(long var1);
}

Example 


public class MyTest8 { 
    public static void main(String[] args){
        Config config=new Config();
        config.useSingleServer().setAddress("redis://192.168.57.120:6379").setPassword("123456");
        RedissonClient client= Redisson.create(config);
 
        RRateLimiter rateLimiter=client.getRateLimiter("rate_limiter");
        rateLimiter.trySetRate(RateType.PER_CLIENT,5,2, RateIntervalUnit.MINUTES); 
        ExecutorService executorService= Executors.newFixedThreadPool(10);
        for (int i=0;i<10;i++){
            executorService.submit(()->{
               try{
                   rateLimiter.acquire();
                   System.out.println(" Thread "+Thread.currentThread().getId()+" Enter the data area: "+System.currentTimeMillis());
               }catch (Exception e){
                   e.printStackTrace();
               }
            });
        }
    }
}

Console output

Thread 49 enters data area: 1574672546522
Thread 55 enters data area: 1574672546522
Thread 56 enters data area: 1574672546526
Thread 50 enters data area: 1574672546523
Thread 48 enters data area: 1574672546523

Thread 51 enters data area: 1574672666627
Thread 53 enters data area: 1574672666627
Thread 54 enters data area: 1574672666627
Thread 57 enters data area: 1574672666628
Thread 52 enters data area: 1574672666628
Explanation: Up to 5 threads are executing in two minutes

Application and Principle of Distributed Current Limiting redission RRateLimiter

Premises:

Recently, the company is doing distributed current limiting in demand, and the current limiting framework for research is probably

1. spring cloud gateway integrates redis current limiting, but it belongs to gateway layer current limiting 2. Ali Sentinel, with powerful functions and monitoring platform 3. srping cloud hystrix belongs to interface layer current limiting and provides two modes: thread pool and semaphore 4. Others: redission, hand code

The actual demand situation belongs to service-side current limiting, and redission is more convenient and flexible to use. The following introduces how to use and principle of redission distributed current limiting:

STEP 1 Use

It is simple to use, as follows 


// 1 ,   Declaration 1 Current limiter 
RRateLimiter rateLimiter = redissonClient.getRateLimiter(key);
 
// 2 ,   Set the rate, 5 Generated in seconds 3 Token 
rateLimiter.trySetRate(RateType.OVERALL, 3, 5, RateIntervalUnit.SECONDS);
 
// 3 Attempting to get 1 Token, obtained to return true
rateLimiter.tryAcquire(1)

2. Principle

1. getRateLimiter 


//  Declaration 1 Current limiter   Name   Call key
redissonClient.getRateLimiter(key)

2. trySetRate

The trySetRate method follows up the implementation of the bottom layer as follows: 


@Override
    public RFuture<Boolean> trySetRateAsync(RateType type, long rate, long rateInterval, RateIntervalUnit unit) {
        return commandExecutor.evalWriteAsync(getName(), 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.<Object>singletonList(getName()), rate, unit.toMillis(rateInterval), type.ordinal());
    }

For example, it is easier to understand:

For example, in the following code, three tokens are generated in 5 seconds, and all instances are shared (RateType. OVERALL all instances are shared, RateType. CLIENT single instance side is shared) 


trySetRate(RateType.OVERALL, 3, 5, RateIntervalUnit.SECONDS);

Then three parameters are set in redis:

hsetnx,key,rate,3
hsetnx,key,interval,5
hsetnx,key,type,0

Then look at the tryAcquire (1) method: the underlying source code is as follows 


private <T> RFuture<T> tryAcquireAsync(RedisCommand<T> command, Long value) {
        return commandExecutor.evalWriteAsync(getName(), LongCodec.INSTANCE, command,
                "local rate = redis.call('hget', KEYS[1], 'rate');"  //1
              + "local interval = redis.call('hget', KEYS[1], 'interval');"  //2
              + "local type = redis.call('hget', KEYS[1], 'type');" //3
              + "assert(rate ~= false and interval ~= false and type ~= false, 'RateLimiter is not initialized')" //4
              
              + "local valueName = KEYS[2];" //5
              + "if type == 1 then "
                  + "valueName = KEYS[3];" //6
              + "end;"
              
              + "local currentValue = redis.call('get', valueName); " //7
              + "if currentValue ~= false then " 
                     + "if tonumber(currentValue) < tonumber(ARGV[1]) then " //8
                         + "return redis.call('pttl', valueName); "
                     + "else "
                         + "redis.call('decrby', valueName, ARGV[1]); " //9
                         + "return nil; "
                     + "end; "
              + "else " //10
                     + "redis.call('set', valueName, rate, 'px', interval); " 
                     + "redis.call('decrby', valueName, ARGV[1]); "
                     + "return nil; "
              + "end;",
                Arrays.<Object>asList(getName(), getValueName(), getClientValueName()), 
                value, commandExecutor.getConnectionManager().getId().toString());
    }

Note lines 1, 2 and 3 get the three values of set in the previous step: rate, interval and type. If these three values are not set, it is directly returned that rateLimiter has not been initialized.

The fifth remark line declares that a variable called valueName has a value of KEYS [2], the corresponding value of KEYS [2] is getValueName () method, and getValueName () returns key set by us in step 1 above; If type=1, indicating global sharing, then the value of valueName is changed to KEYS [3], and the corresponding value of KEYS [3] is getClientValueName (). Look at the source code of getClientValueName (): 


String getClientValueName() {
        return suffixName(getValueName(), commandExecutor.getConnectionManager().getId().toString());
   }

ConnectionManager (). getId () is as follows: 


public enum RateType {
    OVERALL,             // Total current limit for all clients 
    PER_CLIENT;          // Each client calculates traffic separately 
 
    private RateType() {
    }
}
0

This getId () is the UUID generated when each client is initialized, that is, the getId of each client is only 1, which verifies the role of RateType.ALL and RateType.PER_CLIENT in trySetRate methods.

Then look at the 7 standard line and get the value currentValue corresponding to valueName; For the first time, the positive is null, so look at the logic of else in the 10th standard line set valueName 3 px 5, set key=valueName value=3 expiration time to 5 seconds decrby valueName 1, subtract the value of valueName above by 1 Then, if the value returned by the 7th annotation line exists in the second access, the 8th annotation line will go, followed by the following judgment If the current value of valueName is 3, which is less than the number of tokens to be obtained (the input parameter in tryAcquire method), it means that the number of tokens has been used up within the current time (within 5 seconds of the validity period of key), and pttl is returned (the remaining expiration time of key); On the contrary, it means that there are enough tokens in the bucket. After obtaining, the number of tokens in the bucket will be reduced by 1, and this is the end.

Summary:

redission distributed current limiting adopts token bucket idea and fixed time window, trySetRate method sets bucket size, redis key expiration mechanism achieves the purpose of time window, and controls the amount of requests allowed to pass in fixed time window.

Related articles: