Distributed current limiting, everything you want to know is here

Preface

In a high-concurrency system, it is very important to control the flow. When a huge amount of traffic is directly requested to our server, it may cause the interface to become unavailable in a short time. If it is not handled, it may even cause the entire application to become unavailable.

For example, recently there is such a requirement. As a client, I need to produce data to Kafka, and Kafka consumers consume data continuously and request all the consumed data to the web server. Although the load is done (there are 4 web servers), the amount of business data is also huge, and there may be tens of thousands of data generated every second. If the producer produces data directly, it is very likely to drag down the web server.

[[263347]]

For this, it is necessary to do current limiting processing and produce a certain amount of data to Kafka every second, so as to ensure the normal operation of the web to a great extent.

In fact, no matter what scenario you are dealing with, the essence is to reduce traffic and ensure high availability of the application.

Common Algorithms

There are two common algorithms for current limiting:

• Leaky Bucket Algorithm
• Token Bucket Algorithm

The leaky bucket algorithm is relatively simple, which is to put the traffic into the bucket, and the leaky bucket will also flow out at a certain rate. If the traffic is too fast, it will overflow (the leaky bucket does not increase the outflow rate). The overflowing traffic is directly discarded.

As shown in the following figure:

This approach is simple and crude.

Although the leaky bucket algorithm is simple, it cannot cope with actual scenarios, such as a sudden surge in traffic.

At this time, the token bucket algorithm is needed:

The token bucket will put tokens into a fixed-capacity bucket at a constant rate, and take away one or more tokens when traffic comes. If there are no tokens in the bucket, the current request will be discarded or blocked.

In contrast, the token bucket can handle a certain amount of burst traffic.

RateLimiter Implementation

For the code implementation of the token bucket, you can directly use the RateLimiter in the Guava package.

 @Override
 public BaseResponse<UserResVO> getUserByFeignBatch(@RequestBody UserReqVO userReqVO) {
 //Call remote service
 OrderNoReqVO vo = new OrderNoReqVO();
 vo.setReqNo(userReqVO.getReqNo());
 RateLimiter limiter = RateLimiter.create (2.0);
 //Batch call
 for ( int i = 0 ; i< 10 ; i++){
 double acquire = limiter.acquire();
 logger.debug( "Get token successfully!,consumption=" + acquire);
 BaseResponse<OrderNoResVO> orderNo = orderServiceClient.getOrderNo(vo);
 logger.debug( "Remote return:" +JSON.toJSONString(orderNo));
 }
 UserRes userRes = new UserRes() ;
 userRes.setUserId(123);
 userRes.setUserName( "张三" );
 userRes.setReqNo(userReqVO.getReqNo());
 userRes.setCode(StatusEnum.SUCCESS.getCode());
 userRes.setMessage( "Success" );
 return userRes;
 }

See here for details.

The call results are as follows:

From the code, we can see that two tokens are put into the bucket every second, and one token is consumed for each request. Therefore, only two requests can be sent per second. This is indeed the case according to the time in the figure (the return value is the time consumed to obtain this token, which is also about one token every 500ms).

There are several things to note when using RateLimiter:

It allows consumption first and payment later, which means that it can take a few tokens at a time or all the remaining tokens or even more when a request comes in, but the subsequent requests have to pay for the previous request. It needs to wait until the tokens in the bucket are replenished before it can continue to obtain tokens.

Summarize

For a single application, RateLimiter is sufficient. If it is a distributed environment, it can be done with the help of Redis.

Come and demonstrate.

Current limiting is adopted in the interface provided by the Order application. First, the bean of the current limiting tool is configured:

 @Configuration
 public class RedisLimitConfig {
 @Value( "${redis.limit}" )
 private int limit;
 @Autowired
 private JedisConnectionFactory jedisConnectionFactory;
 @Bean
 public RedisLimit build() {
 RedisClusterConnection clusterConnection = jedisConnectionFactory.getClusterConnection();
 JedisCluster jedisCluster = (JedisCluster) clusterConnection.getNativeConnection();
 RedisLimit redisLimit = new RedisLimit.Builder<>(jedisCluster)
 .limit(limit)
 .build();
 return redisLimit;
 }
 }

Then use the component in the Controller:

 @Autowired
 private RedisLimit redisLimit;
 @Override
 @CheckReqNo
 public BaseResponse<OrderNoResVO> getOrderNo(@RequestBody OrderNoReqVO orderNoReq) {
 BaseResponse<OrderNoResVO> res = new BaseResponse();
 //Current limiting
 boolean limit = redisLimit.limit();
 if (!limit){
 res.setCode(StatusEnum.REQUEST_LIMIT.getCode());
 res.setMessage(StatusEnum.REQUEST_LIMIT.getMessage());
 return res ;
 }
 res.setReqNo(orderNoReq.getReqNo());
 if ( null == orderNoReq.getAppId()){
 throw new SBCException(StatusEnum.FAIL);
 }
 OrderNoResVO orderNoRes = new OrderNoResVO();
 orderNoRes.setOrderId(DateUtil.getLongTime());
 res.setCode(StatusEnum.SUCCESS.getCode());
 res.setMessage(StatusEnum.SUCCESS.getMessage());
 res.setDataBody(orderNoRes);
 return res ;
 }

For ease of use, annotations are also provided:

 @Override
 @ControllerLimit
 public BaseResponse<OrderNoResVO> getOrderNoLimit(@RequestBody OrderNoReqVO orderNoReq) {
 BaseResponse<OrderNoResVO> res = new BaseResponse();
 // Business logic
 return res ;
 }

This annotation intercepts the http request and returns directly when the request reaches the threshold.

The normal method can also be used:

 @CommonLimit
 public void doSomething(){}

An exception will be thrown when the call threshold is reached.

In order to simulate concurrency, 10 threads are opened in the User application to call the Order interface (the current limit is 5 times) (professional concurrency testing tools such as JMeter can also be used).

 @Override
 public BaseResponse<UserResVO> getUserByFeign(@RequestBody UserReqVO userReq) {
 //Call remote service
 OrderNoReqVO vo = new OrderNoReqVO();
 vo.setAppId(1L);
 vo.setReqNo(userReq.getReqNo());
 for ( int i = 0; i < 10; i++) {
 executorService.execute (new Worker(vo, orderServiceClient));
 }
 UserRes userRes = new UserRes();
 userRes.setUserId(123);
 userRes.setUserName( "张三" );
 userRes.setReqNo(userReq.getReqNo());
 userRes.setCode(StatusEnum.SUCCESS.getCode());
 userRes.setMessage( "Success" );
 return userRes;
 }
 private static class Worker implements Runnable {
 private OrderNoReqVO vo;
 private OrderServiceClient orderServiceClient;
 public Worker(OrderNoReqVO vo, OrderServiceClient orderServiceClient) {
 this.vo = vo;
 this.orderServiceClient = orderServiceClient;
 }
 @Override
 public void run() {
 BaseResponse<OrderNoResVO> orderNo = orderServiceClient.getOrderNoCommonLimit(vo);
 logger.info( "Remote return:" + JSON.toJSONString(orderNo));
 }
 }

In order to verify the distributed effect, two Order applications are started.

The effect is as follows:

Implementation principle

The implementation principle is actually very simple. Since we want to achieve the effect of distributed global current limiting, we naturally need a third-party component to record the number of requests.

Redis is very suitable for such scenarios.

• The current time (accurate to seconds) is written into Redis as a key for each request, and the timeout is set to 2 seconds. Redis increments the value of the key.
• Returns an error when the threshold is reached.
• The operation of writing to Redis is completed using Lua scripts, and the single-threaded mechanism of Redis can ensure the atomicity of each Redis request.

The Lua script is as follows:

--lua subscript starts from 1-- Current limiting keylocal key = KEYS[1]-- Current limiting sizelocal limit = tonumber(ARGV[1])-- Get current flow sizelocal curentLimit = tonumber(redis.call('get', key) or "0")if curentLimit + 1 > limit then -- Return return 0 if the current limiting size is reached;else -- The threshold value + 1 is not reached redis.call("INCRBY", key, 1) redis.call("EXPIRE", key, 2) return curentLimit + 1end

Calling logic in Java:

 --lua subscript starts from 1  
 -- Current limiting key  
 local   key = KEYS[1]
 -- Current limit size  
 local limit = tonumber(ARGV[1])
 -- Get the current flow size  
 local curentLimit = tonumber(redis.call( 'get' , key ) or   "0" )
 if curentLimit + 1 > limit then  
 -- Reach the current limit size return  
 return 0;
 else  
 -- The threshold value + 1 is not reached  
 redis.call( "INCRBY" , key , 1)
 redis.call( "EXPIRE" , key , 2)
 return curentLimit + 1
 end  

Therefore, you only need to call this method where current limiting is needed to judge the return value to achieve the purpose of current limiting.

Of course, this is just a crude counter made using Redis. If you want to implement a token bucket algorithm similar to the one mentioned above, you can implement it yourself based on Lua.

Builder

When designing this component, we tried to provide users with a clear, readable, and error-prone API.

For example, in the first step, how to build a current limiting object.

The most common way is of course the constructor. If there are multiple domains, you can use the overlapping constructor method:

 public A(){}
 public A( int a){}
 public A( int a, int b){}

The disadvantages are also obvious: if there are too many parameters, it will be difficult to read. Even if the parameter types are consistent and the client reverses the order, it will not cause a warning, resulting in unpredictable results.

The second solution is to use the JavaBean mode and construct it using the setter method:

 A a = new A();
 a.setA(a);
 a.setB(b);

This approach is clear and easy to read, but it is easy to put the object in an inconsistent state and make the object thread-unsafe.

So here we use the third way to create objects, the builder:

 public class RedisLimit {
 private JedisCommands jedis;
 private int limit = 200;
 private static final int FAIL_CODE = 0;
 /**
 * lua script
 */
 private String script;
 private RedisLimit(Builder builder) {
 this.limit = builder.limit;
 this.jedis = builder.jedis;
 buildScript();
 }
 /**
 * limit traffic
 * @return if true  
 */
 public boolean limit() {
 String key = String.valueOf(System.currentTimeMillis() / 1000);
 Object result = null ;
 if (jedis instanceof Jedis) {
 result = ((Jedis) this.jedis).eval(script, Collections.singletonList( key ), Collections.singletonList(String.valueOf(limit)));
 } else if (jedis instanceof JedisCluster) {
 result = ((JedisCluster) this.jedis).eval(script, Collections.singletonList( key ), Collections.singletonList(String.valueOf(limit)));
 } else {
 //throw new RuntimeException( "instance is error" ) ;
 return   false ;
 }
 if (FAIL_CODE != (Long) result) {
 return   true ;
 } else {
 return   false ;
 }
 }
 /**
 * read lua script
 */
 private void buildScript() {
 script = ScriptUtil.getScript( "limit.lua" );
 }
 /**
 * the builder
 * @param <T>
 */
 public   static class Builder<T extends JedisCommands>{
 private T jedis = null ;
 private int limit = 200;
 public Builder(T jedis){
 this.jedis = jedis;
 }
 public Builder limit( int limit){
 this.limit = limit;
 return this;
 }
 public RedisLimit build(){
 return new RedisLimit(this);
 }
 }
 }

So when the client uses it:

 RedisLimit redisLimit = new RedisLimit.Builder<>(jedisCluster)
 .limit(limit)
 .build();

It is much simpler and more direct, and avoids breaking the creation process into multiple sub-steps.

This is useful when there are multiple constructor parameters, but they are not required fields.

Therefore, the distributed lock builder method is also updated:

https://github.com/crossoverJie/distributed-redis-tool#features

API

As can be seen from the above, the usage process is to call the limit method.

 //Current limiting
 boolean limit = redisLimit.limit();
 if (!limit){
 //Specific current limiting logic
 }

In order to reduce intrusion and simplify the client, two annotation methods are provided.

@ControllerLimit

This annotation can be used in interfaces modified by @RequestMapping and will provide a current limiting response after current limiting.

The implementation is as follows:

 @Component
 public class WebIntercept extends WebMvcConfigurerAdapter {
 private static Logger logger = LoggerFactory.getLogger(WebIntercept.class);
 @Autowired
 private RedisLimit redisLimit;
 @Override
 public void addInterceptors(InterceptorRegistry registry) {
 registry.addInterceptor(new CustomInterceptor())
 .addPathPatterns( "/**" );
 }
 private class CustomInterceptor extends HandlerInterceptorAdapter {
 @Override
 public boolean preHandle(HttpServletRequest request, HttpServletResponse response,
 Object handler) throws Exception {
 if (redisLimit == null ) {
 throw new NullPointerException( "redisLimit is null" );
 }
 if (handler instanceof HandlerMethod) {
 HandlerMethod method = (HandlerMethod) handler;
 ControllerLimit annotation = method.getMethodAnnotation(ControllerLimit.class);
 if (annotation == null ) {
 //skip
 return   true ;
 }
 boolean limit = redisLimit.limit();
 if (!limit) {
 logger.warn( "request has bean limit" );
 response.sendError(500, "request limit" );
 return   false ;
 }
 }
 return   true ;
 }
 }
 }

In fact, it implements the interceptor in SpringMVC, and determines whether annotations are used during the interception process, thereby calling the current limiting logic.

The premise is that the application needs to scan the class and let Spring manage it.

 @ComponentScan(value = "com.crossoverjie.distributed.intercept" )

@CommonLimit

Of course, it can also be used in ordinary methods. The implementation principle is Spring AOP (SpringMVC's interceptor is essentially AOP).

 @Aspect
 @Component
 @EnableAspectJAutoProxy(proxyTargetClass = true )
 public class CommonAspect {
 private static Logger logger = LoggerFactory.getLogger(CommonAspect.class);
 @Autowired
 private RedisLimit redisLimit;
 @Pointcut( "@annotation(com.crossoverjie.distributed.annotation.CommonLimit)" )
 private void check (){}
 @Before( "check()" )
 public void before(JoinPoint joinPoint) throws Exception {
 if (redisLimit == null ) {
 throw new NullPointerException( "redisLimit is null" );
 }
 boolean limit = redisLimit.limit();
 if (!limit) {
 logger.warn( "request has bean limit" );
 throw new RuntimeException( "request has bean limit" ) ;
 }
 }
 }

It's very simple, and the current limiting is also called during the interception process.

Of course, you also have to scan the package when using it:

 @ComponentScan(value = "com.crossoverjie.distributed.intercept" )

Summarize

Current limiting is a powerful tool for protecting applications in a high-concurrency and high-traffic system. There are many mature solutions. I hope to provide some ideas for friends who are just beginning to understand this area.