How to implement single-machine current limiting in java

When to use rate limiting:

For example, if you hope that the QPS of your application will not exceed 1000, then after RateLimiter sets a rate of 1000, it will throw 1000 tokens into the bucket every second. RateLimiter often Used to limit the access rate to some physical resources or logical resources.

Introduction:

For the current limit of the stand-alone version, you can use Google's open source Guava project. This project provides some core libraries used by Google in Java projects, including collections (Collections) and cache. (Caching), concurrent programming library (Concurrency), common annotations (Common annotations), String operations, and many very practical functions in I/O operations.

This project also includes the current limiting function, which is implemented based on the token bucket algorithm.

Provides two current limiting strategies:

● Smooth burst current limiting (SmoothBursty)
● Smooth preheating current limiting (SmoothWarmingUp) implementation.

Dependencies:

<dependency>
  <groupId>com.google.guava</groupId>
  <artifactId>guava</artifactId>
  <version>29.0-jre</version>
</dependency>

Method description:

Simulation scenario (example):

Scenario 1:

When we hope that the number of visits to a certain interface will not exceed 10 times per second

package org.xhs.test;

import org.apache.curator.shaded.com.google.common.util.concurrent.RateLimiter;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;


/**
 * @Author: hu.chen
 * @Description:
 **/
public class Test {

    /**
     * 存储接口名和令牌生成器的对应关系
     */
   private static Map<String, RateLimiter> interfaces = new ConcurrentHashMap<>();
   
    /**
     * 线程池
     */
    private static ExecutorService threadPool = new ThreadPoolExecutor(10,15,3,TimeUnit.SECONDS,new ArrayBlockingQueue<>(100));

    public static void main(String[] args) throws InterruptedException {

        List<UserRequest> tasks = new ArrayList<UserRequest>();
        // 准备工作，先初始化 10个线程（用户），这10个用户同时访问一个接口
        for (int i = 1; i <= 12; i++) {
            String ip = "127.0.0." + i;
            String userName="chenhu_"+i;
            String interfaceName="user/find_";
            tasks.add(new UserRequest(ip,userName,interfaceName));
        }


        // 先初始化好令牌生成器
        for (UserRequest request : tasks) {
            // 根据接口名限流
            RateLimiter rateLimiter = interfaces.get(request.getInterfaceName());

            if(rateLimiter==null){
                // 创建一个令牌生成器，每秒产生10个令牌
                synchronized (interfaces) {
                    if(rateLimiter==null) {
                        rateLimiter = RateLimiter.create(10);
                        // 将这个令牌生成器和具体的接口进行绑定
                        interfaces.put(request.getInterfaceName(),rateLimiter);
                    }
                }
            }
        }

        // 休眠一秒，让令牌生成器先生成令牌
        Thread.sleep(1000);

        for (UserRequest request : tasks) {
            // 根据接口名限流
            RateLimiter rateLimiter = interfaces.get(request.getInterfaceName());


            // 获取令牌桶中一个令牌，如果获取不到，则等待 timeout 时间，如果还获取不到，则返回false，反之则返回true
            // timeout设置为0，表示不等待
            if(rateLimiter.tryAcquire(1,0,TimeUnit.SECONDS)){

                // 得到令牌，处理请求
                threadPool.execute(()->{
                    System.err.println("接口："+request.getInterfaceName()+" 访问还未达到上限，"+request.getUserName()+"可以访问");
                });
            }else {
                // 已经等待了10秒还获取不到令牌，进行其他业务处理
                System.err.println("当前时间访问失败，"+request.getUserName()+"无法获取令牌");
            }
        }
    }

    private static class UserRequest {
        /**
         * 请求用户ip
         */
        private String ip;

        /**
         * 用户名
         */
        private String userName;

        /**
         * 请求的接口名
         */
        private String interfaceName;
        public UserRequest(String ip, String userName, String interfaceName) {
            this.ip = ip;
            this.userName = userName;
            this.interfaceName = interfaceName;
        }
        public String getIp() {return ip;}

        public String getUserName() { return userName;}
        public String getInterfaceName() {return interfaceName;}
    }
}

Scenario 2:

When we hope that a certain user or IP will have no more than 10 visits per second

package org.xhs.test;

import org.apache.curator.shaded.com.google.common.util.concurrent.RateLimiter;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;


/**
 * @Author: hu.chen
 * @Description:
 **/
public class Test {

    /**
     * 存储用户名和令牌生成器的对应关系
     */
   private static Map<String, RateLimiter> interfaces = new ConcurrentHashMap<>();

    /**
     * 线程池
     */
    private static ExecutorService threadPool = new ThreadPoolExecutor(10,15,3,TimeUnit.SECONDS,new ArrayBlockingQueue<>(100));

    public static void main(String[] args) throws InterruptedException {

        List<UserRequest> tasks = new ArrayList<UserRequest>();
        // 准备工作，先初始化 10个线程（用户），这10个用户同时访问一个接口
        for (int i = 1; i <= 12; i++) {
            String ip = "127.0.0." + i;
            String userName="chenhu_";
            String interfaceName="user/find_"+i;
            tasks.add(new UserRequest(ip,userName,interfaceName));
        }


        // 先初始化好令牌生成器
        for (UserRequest request : tasks) {
            // 根据接口名限流
            RateLimiter rateLimiter = interfaces.get(request.getUserName());

            if(rateLimiter==null){
                // 创建一个令牌生成器，每秒产生5个令牌
                synchronized (interfaces) {
                    if(rateLimiter==null) {
                        rateLimiter = RateLimiter.create(10);
                        // 将这个令牌生成器和具体的接口进行绑定
                        interfaces.put(request.getUserName(),rateLimiter);
                    }
                }
            }
        }

        // 休眠一秒，让令牌生成器先生成令牌
        Thread.sleep(1000);

        for (UserRequest request : tasks) {
            // 根据接口名限流
            RateLimiter rateLimiter = interfaces.get(request.getUserName());


            // 获取令牌桶中一个令牌，如果获取不到，则等待 timeout 时间，如果还获取不到，则返回false，反之则返回true
            // timeout设置为0，表示不等待
            if(rateLimiter.tryAcquire(1,0,TimeUnit.SECONDS)){

                // 得到令牌，处理请求
                threadPool.execute(()->{
                    System.err.println("用户："+request.getUserName()+" 当前时间访问次数还未达到上限，可以访问");
                });
            }else {
                // 已经等待了10秒还获取不到令牌，进行其他业务处理
                System.err.println("当前时间访问失败，"+request.getUserName()+"无法获取令牌");
            }
        }
    }

    private static class UserRequest {
        /**
         * 请求用户ip
         */
        private String ip;

        /**
         * 用户名
         */
        private String userName;

        /**
         * 请求的接口名
         */
        private String interfaceName;
        public UserRequest(String ip, String userName, String interfaceName) {
            this.ip = ip;
            this.userName = userName;
            this.interfaceName = interfaceName;
        }
        public String getIp() {return ip;}

        public String getUserName() { return userName;}
        public String getInterfaceName() {return interfaceName;}
    }
}

The above is the detailed content of How to implement single-machine current limiting in java. For more information, please follow other related articles on the PHP Chinese website!