Method for realizing distributed application current limiting based on Redis

The purpose of current limiting is to protect the system by limiting the speed of concurrent access/requests or limiting the speed of requests within a time window. Once the limiting rate is reached, service can be denied.

A few days ago, I read a plan to use Guawa to achieve single-application stream limiting. Reference "Redis in action" to implement a Jedis version, all of which are business-level restrictions. Common current limiting strategies in actual scenarios:

Nginx access layer current limit

According to certain rules, such as account number, IP, system call logic, etc., to limit current at the Nginx level

Business application system current limit

Controlling traffic through business code This traffic can be called a semaphore, which can be understood as a lock, which can limit how many processes a resource can be accessed simultaneously.

Code implementation

 import redis.clients.jedis.Jedis;import redis.clients.jedis.Transaction;import redis.clients.jedis.ZParams;import java.util.List;import java.util.UUID;/** * @email [email protected] * @data 2017-08 */public class RedisRateLimiter { private static final String BUCKET = "BUCKET"; private static final String BUCKET_COUNT = "BUCKET_COUNT"; private static final String BUCKET_MONITOR = "BUCKET_MONITOR"; static String acquireTokenFromBucket( Jedis jedis, int limit, long timeout) { String identifier = UUID.randomUUID().toString(); long now = System.currentTimeMillis(); Transaction transaction = jedis.multi(); //Delete semaphore transaction.zremrangeByScore(BUCKET_MONITOR.getBytes(), "-inf".getBytes(), String.valueOf(now - timeout).getBytes()); ZParams params = new ZParams(); params.weightsByDouble(1.0,0.0); transaction.zinterstore(BUCKET, params, BUCKET, BUCKET_MONITOR); //Counter self-increment transaction.incr(BUCKET_COUNT); List<Object> results = transaction.exec(); long counter = (Long) results.get(results.size() - 1); transaction = jedis.multi(); transaction.zadd(BUCKET_MONITOR, now, identifier); transaction.zadd(BUCKET, counter, identifier); transaction.zrank(BUCKET, identifier); results = transaction.exec(); //Get the ranking to determine whether the request has obtained the semaphore long rank = (Long) results.get(results.size() - 1); if (rank < limit) { return identifier; } else {//No semaphore was obtained, put into redis before cleaning transaction = jedis.multi(); transaction.zrem(BUCKET_MONITOR, identifier); transaction.zrem(BUCKET, identifier); transaction.exec(); } return null; }}

Call

Test interface call

 @GetMapping("/")public void index(HttpServletResponse response) throws IOException { Jedis jedis = jedisPool.getResource(); String token = RedisRateLimiter.acquireTokenFromBucket(jedis, LIMIT, TIMEOUT); if (token == null) { response.sendError(500); }else{ //TODO Your business logic} jedisPool.returnResource(jedis);}

optimization

Optimize code with interceptor + annotation

Interceptor

 @Configurationstatic class WebMvcConfigurer extends WebMvcConfigurerAdapter { private Logger logger = LoggerFactory.getLogger(WebMvcConfigurer.class); @Autowired private JedisPool jedisPool; public void addInterceptors(InterceptorRegistry registry) { registry.addInterceptor(new HandlerInterceptorAdapter() { public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception { HandlerMethod handlerMethod = (HandlerMethod) handler; Method method = handlerMethod.getMethod(); RateLimiter rateLimiter = method.getAnnotation(RateLimiter.class); if (rateLimiter != null){ int limit = rateLimiter.limit(); int timeout = rateLimiter.timeout(); Jedis jedis = jedisPool.getResource(); String token = RedisRateLimiter.acquireTokenFromBucket(jedis, limit, timeout); if (token == null) { response.sendError(500); return false; } logger.debug("token -> {}",token); jedis.close(); } return true; } }).addPathPatterns("/*"); }}

Definition annotation

 /** * @email [email protected] * @data 2017-08 * Current limit annotation*/@Target(ElementType.METHOD)@Retention(RetentionPolicy.RUNTIME)@Documentedpublic @interface RateLimiter { int limit() default 5; int timeout() default 1000;}

use

 @RateLimiter(limit = 2, timeout = 5000)@GetMapping("/test")public void test() {}

Concurrent testing

Tools: apache-jmeter-3.2

Note: The interface that has not obtained the semaphore returns 500, status is red, the interface that has obtained the semaphore returns 200, status is green.

When the limit request semaphore is 2, 5 threads are sent:

When the limit request semaphore is 5, 10 threads are sent:

material

Implementation based on reids + lua

Summarize

1. For semaphore operations, use transaction operations.
2. Do not use timestamps as the sorting score of semaphores, because in a distributed environment, there will be unfair semaphores due to the time difference between each node.
3. You can use this code to extract the @rateLimiter annotation, and then use it on the method.
4. For flow control of different interfaces, you can refer to the source code RedisRateLimiterPlus, which is nothing more than generating a monitoring parameter for each interface.
5. Source code: boding1-pig-cloud-jb51.rar

The above is all the content of this article. I hope it will be helpful to everyone's learning and I hope everyone will support Wulin.com more.