Java Concurrent Programming Introduction (11) stream limiting scenarios and Spring stream limiting implementation

Java geek

1. Traffic limiting scenario

Traffic limiting is based on the usage of hardware resources, including CPU, memory, and I/O. For example, a report service consumes a lot of memory. If the number of concurrent requests increases, the entire application will be slowed down, or even the memory will overflow and the application will die.

Current limit applicable to dynamic increase of resource, have been pooling resources does not necessarily need to current limit, such as database connection pool, it is has certain resources, pool size is fixed (even to dynamically scale pool size), such a scenario does not need by current limit, as long as can do if pool link has been used up, is unable to get a new connection can be.

Therefore, the prerequisites for using current limiting are as follows: 1. Prevent adverse effects caused by resource overload. 2. The resources used will increase dynamically, such as requests from a site.

Second, Spring in the implementation of traffic limiting

I. Demand for current limiting

1. Limit traffic for controllers only. 2. You can match urls based on regular expressions to limit traffic. 4. You can define multiple traffic limiting rules, and each rule has a different maximum traffic

II. Related class structure

III. Show me code

1.AcquireResult.java

public class AcquireResult {

    /** Obtain pass successfully */
    public static final int ACQUIRE_SUCCESS = 0;

    /** Failed to get pass */
    public static final int ACQUIRE_FAILED = 1;

    /** Do not need to obtain a pass */
    public static final int ACQUIRE_NONEED = 2;

    /** Get pass result */
    private int result;

    /** Number of passes available */
    private int availablePermits;

    public int getResult(a) {
        return result;
    }

    public void setResult(int result) {
        this.result = result;
    }

    public int getAvailablePermits(a) {
        return availablePermits;
    }

    public void setAvailablePermits(int availablePermits) {
        this.availablePermits = availablePermits; }}Copy the code

2.LimiteRule.java

/ * * *@ClassName LimiteRule
 * @Description TODO
 * @AuthorSonorous one leaf *@Date2019/10/4 prepare *@Version1.0 * * * / javashizhan.com
public class LimiteRule {

    /** semaphore */
    private final Semaphore sema;

    /** Request URL matching rule */
    private final String pattern;

    /** Maximum number of concurrent requests */
    private final int maxConcurrent;

    public LimiteRule(String pattern, int maxConcurrent) {
        this.sema = new Semaphore(maxConcurrent);
        this.pattern = pattern;
        this.maxConcurrent = maxConcurrent;
    }

    /** * Get a pass. The use of the synchronized keyword is used in order to appear to decrease or increase the number of passes available for printing@paramUrlPath Request Url *@return0- Obtain success, 1- no pass was obtained, 2- no pass required */
    public synchronized AcquireResult tryAcquire(String urlPath) {

        AcquireResult acquireResult = new AcquireResult();
        acquireResult.setAvailablePermits(this.sema.availablePermits());

        try {
            // The Url requests matching rules to obtain the pass
            if (Pattern.matches(pattern, urlPath)) {

                boolean acquire = this.sema.tryAcquire(50, TimeUnit.MILLISECONDS);

                if (acquire) {
                    acquireResult.setResult(AcquireResult.ACQUIRE_SUCCESS);
                    print(urlPath);
                } else{ acquireResult.setResult(AcquireResult.ACQUIRE_FAILED); }}else{ acquireResult.setResult(AcquireResult.ACQUIRE_NONEED); }}catch (InterruptedException e) {
            e.printStackTrace();
        }

        return acquireResult;
    }

    /** * release passes. The synchronized keyword is used to print the number of passes that appear to decrease or increase one by one
    public synchronized void release(a) {
        this.sema.release();
        print(null);
    }

    /** * get maximum concurrency *@return* /
    public int getMaxConcurrent(a) {
        return this.maxConcurrent;
    }

    /** * get the matching expression *@return* /
    public String getPattern(a) {
        return this.pattern;
    }

    /** * Print log *@param urlPath
     */
    private void print(String urlPath) {
        StringBuffer buffer = new StringBuffer();
        buffer.append("Pattern: ").append(pattern).append(",");
        if (null! = urlPath) { buffer.append("urlPath: ").append(urlPath).append(",");
        }
        buffer.append("Available Permits:").append(this.sema.availablePermits()); System.out.println(buffer.toString()); }}Copy the code

3.CurrentLimiter.java

/ * * *@ClassName CurrentLimiter
 * @Description TODO
 * @AuthorSonorous one leaf *@Date2019/10/4 prepare *@Version1.0 * * * / javashizhan.com
public class CurrentLimiter {

    /** A local thread variable that stores the pass obtained by a request, separates it from other concurrent requests, and releases the pass obtained by this request after the controller finishes executing it */
    private static ThreadLocal<Vector<LimiteRule>> localAcquiredLimiteRules = new ThreadLocal<Vector<LimiteRule>>();

    /** All finite flow rules */
    private static Vector<LimiteRule> allLimiteRules = new Vector<LimiteRule>();

    /** Private constructor to avoid instantiating */
    private CurrentLimiter(a) {}

    /** * Add a flow limiting rule, which is added at spring startup and does not require locking. If added dynamically at run, it requires locking *@param rule
     */
    public static void addRule(LimiteRule rule) {
        printRule(rule);
        allLimiteRules.add(rule);
    }

    /** * Obtain the traffic communication certificate, all traffic rules must be obtained before passing, if one cannot be obtained, throw an exception * multithreaded concurrent, need to lock *@param urlPath
     */
    public static void tryAcquire(String urlPath) throws Exception {
        // Finite stream rules are handled
        if (allLimiteRules.size() > 0) {

            // Traffic rules that can get passes are saved and released after the Controller executes them
            Vector<LimiteRule> acquiredLimitRules = new Vector<LimiteRule>();

            for(LimiteRule rule:allLimiteRules) {
                // Get a pass
                AcquireResult acquireResult = rule.tryAcquire(urlPath);

                if (acquireResult.getResult() == AcquireResult.ACQUIRE_SUCCESS) {
                    acquiredLimitRules.add(rule);
                    // Get the traffic rule from the pass and add it to the local thread variable
                    localAcquiredLimiteRules.set(acquiredLimitRules);

                } else if (acquireResult.getResult() == AcquireResult.ACQUIRE_FAILED) {
                    // Throw an exception if the pass cannot be obtained
                    StringBuffer buffer = new StringBuffer();
                    buffer.append("The request [").append(urlPath).append("] exceeds maximum traffic limit, the limit is ").append(rule.getMaxConcurrent())
                            .append(", available permit is").append(acquireResult.getAvailablePermits()).append(".");

                    System.out.println(buffer);
                    throw new Exception(buffer.toString());

                } else {
                    StringBuffer buffer = new StringBuffer();
                    buffer.append("This path does not match the limit rule, path is [").append(urlPath)
                            .append("], pattern is [").append(rule.getPattern()).append("]."); System.out.println(buffer.toString()); }}}}/** * Release the obtained pass. Drop the call after the controller has finished executing (also needed to throw an exception) */
    public static void release(a) {
        Vector<LimiteRule> acquiredLimitRules = localAcquiredLimiteRules.get();
        if (null! = acquiredLimitRules && acquiredLimitRules.size() >0) {
            acquiredLimitRules.forEach(rule->{
                rule.release();
            });
        }

        // Destory local thread variable to avoid memory leaks
        localAcquiredLimiteRules.remove();
    }

    /** * Print traffic limiting rule information *@param rule
     */
    private static void printRule(LimiteRule rule) {
        StringBuffer buffer = new StringBuffer();
        buffer.append("Add Limit Rule, Max Concurrent: ").append(rule.getMaxConcurrent())
                .append(", Pattern: ").append(rule.getPattern()); System.out.println(buffer.toString()); }}Copy the code

4.CurrentLimiteAspect.java

/ * * *@ClassName CurrentLimiteAspect
 * @Description TODO
 * @AuthorSonorous one leaf *@Date 2019/10/4 20:15
 * @Version1.0 * * * / javashizhan.com
@Aspect
@Component
public class CurrentLimiteAspect {

    /** * Intercept the controller and change the path */
    @Pointcut("execution(* com.javashizhan.controller.. * (..) )")
    public void controller(a) {}@Before("controller()")
    public void controller(JoinPoint point) throws Exception {
        HttpServletRequest request = ((ServletRequestAttributes) RequestContextHolder.getRequestAttributes()).getRequest();
        // Get the pass, urlPath format is: /limit
        CurrentLimiter.tryAcquire(request.getRequestURI());
    }

    /** * This intercepting method will be called even if the controller throws an exception@param joinPoint
     */
    @After("controller()")
    public void after(JoinPoint joinPoint) {
        // Release the obtained passCurrentLimiter.release(); }}Copy the code

5.Application.java

@SpringBootApplication
public class Application {

    public static void main(String[] args) {
        new SpringApplicationBuilder(Application.class).run(args);

        // Add a traffic limiting rule
        LimiteRule rule = new LimiteRule("/limit".4); CurrentLimiter.addRule(rule); }}Copy the code

IV, validation,

Two pits encountered by the test validation: 1. Manually discovered that the controller is serial through the browser refresh request 2. The tests are serial even if the number of concurrent tests is set, as shown in the figure below:

Baidu fruitless, can only write their own code to verify, the code is as follows:

/ * * *@ClassName CurrentLimiteTest
 * @DescriptionVerify the current limiter *@AuthorSonorous one leaf *@Date 2019/10/5 0:51
 * @Version1.0 * * * / javashizhan.com
public class CurrentLimiteTest {

    public static void main(String[] args) {
        final String limitUrlPath = "http://localhost:8080/limit";
        final String noLimitUrlPath = "http://localhost:8080/nolimit";

        // Stream limiting test
        test(limitUrlPath);

        // Sleep for a while, wait for the last batch of threads to finish execution, easy to view the log
        sleep(5000);

        // Unlimited stream tests
        test(noLimitUrlPath);

    }

    private static void test(String urlPath) {
        Thread[] requesters = new Thread[10];

        for (int i = 0; i < requesters.length; i++) {
            requesters[i] = new Thread(newRequester(urlPath)); requesters[i].start(); }}private static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch(InterruptedException e) { e.printStackTrace(); }}}class Requester implements Runnable {

    private final String urlPath;
    private final RestTemplate restTemplate = new RestTemplate();

    public Requester(String urlPath) {
        this.urlPath = urlPath;
    }

    @Override
    public void run(a) {
        String response = restTemplate.getForEntity(urlPath, String.class).getBody();
        System.out.println("response: "+ response); }}Copy the code

The output log is as follows:

Pattern: /limit, urlPath: /limit, Available Permits:3
Pattern: /limit, urlPath: /limit, Available Permits:2
Pattern: /limit, urlPath: /limit, Available Permits:1
Pattern: /limit, urlPath: /limit, Available Permits:0
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
Pattern: /limit, Available Permits:1
Pattern: /limit, Available Permits:2
Pattern: /limit, Available Permits:3
Pattern: /limit, Available Permits:4
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
This path does not match the limit rule, path is [/nolimit] pattern is [/limit].
Copy the code

You can see that the log output information is as follows: 1. The maximum concurrency of the first test URL is 4, and 10 concurrent requests are made at a time. After 4 of them obtain passes, the remaining 6 fail to obtain passes. 2. The 4 requests that get the pass release the pass after the controller finishes executing it. 3. The second test URL did not limit the concurrency, and all 10 requests were successfully executed.

At this point, the current limiter verification is successful.

Note: Removed after synchronization lock (the synchronized keyword), print log like this, you can see the available passport number is not increasing or decreasing, but this does not mean the logic is not correct, this is because the semaphore supports multiple threads into the critical region, before printing, may have reduced the multiple passes, another thread of execution does not necessarily end of the first first, So the number of available certificates is not increasing or decreasing. All the semaphore guarantees is that for every pass you use, you lose one.

Pattern: /limit, urlPath: /limit, Available Permits:2
Pattern: /limit, urlPath: /limit, Available Permits:1
Pattern: /limit, urlPath: /limit, Available Permits:0
Pattern: /limit, urlPath: /limit, Available Permits:2
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
The request [/limit] exceeds maximum traffic limit, the limit is 4, available permit is0.
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
Pattern: /limit, Available Permits:2
Pattern: /limit, Available Permits:4
Pattern: /limit, Available Permits:2
Pattern: /limit, Available Permits:3
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
This path does not match the limit rule, path is [/nolimit], pattern is [/limit].
Copy the code

end.

<– read left mark, left point like!