令牌桶算法限流

 1 package com.cjs.example;
 2 
 3 import com.google.common.util.concurrent.RateLimiter;
 4  org.springframework.web.bind.annotation.RequestMapping;
 5  org.springframework.web.bind.annotation.RestController;
 6 
 7  java.text.SimpleDateFormat;
 8  java.util.Date;
 9 
10 @RestController
11 public class HelloController {
12 
13     private static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
14 
15     final RateLimiter rateLimiter = RateLimiter.create(216 
17     /**
18      * tryAcquire尝试获取permit，默认超时时间是0，意思是拿不到就立即返回false
19      */
20     @RequestMapping("/sayHello")
21     public String sayHello() {
22         if (rateLimiter.tryAcquire()) { //  一次拿1个
23             System.out.println(sdf.format(new Date()));
24             try {
25                 Thread.sleep(50026             } catch (InterruptedException e) {
27                 e.printStackTrace();
28             }
29         }else30             System.out.println("limit"31         }
32         return "hello";
33     }
34 
35     36      * acquire拿不到就等待，拿到为止
37      38     @RequestMapping("/sayHi"39      String sayHi() {
40         rateLimiter.acquire(5);   一次拿5个
41         System.out.println(sdf.format(42         return "hi"43 44 
45 }

final RateLimiter rateLimiter = RateLimiter.create(2.0);  rate is "2 permits per second"
void submitTasks(List<Runnable> tasks,Executor executor) {
    for (Runnable task : tasks) {
        rateLimiter.acquire();  may wait
        executor.execute(task);
    }
}

abstract  RateLimiter {

    /**
     * 用给定的吞吐量（“permits per second”）创建一个RateLimiter。
     * 通常是QPS
     */
    static RateLimiter create(double permitsPerSecond) {
        return create(permitsPerSecond,SleepingStopwatch.createFromSystemTimer());
    }
    
     permitsPerSecond,SleepingStopwatch stopwatch) {
        RateLimiter rateLimiter = new SmoothBursty(stopwatch,1.0 /* maxBurstSeconds */);
        rateLimiter.setRate(permitsPerSecond);
         rateLimiter;
    }
    
    
     * 用给定的吞吐量(QPS)和一个预热期创建一个RateLimiter
     double permitsPerSecond,long warmupPeriod,TimeUnit unit) {
        checkArgument(warmupPeriod >= 0,"warmupPeriod must not be negative: %s",warmupPeriod);
        return create(permitsPerSecond,warmupPeriod,unit,3.0static RateLimiter create(
             coldFactor,1)"> SmoothWarmingUp(stopwatch,coldFactor);
        rateLimiter.setRate(permitsPerSecond);
         rateLimiter;
    }

    final SleepingStopwatch stopwatch;

        锁
    volatile Object mutexDoNotUseDirectly;

    private Object mutex() {
        Object mutex = mutexDoNotUseDirectly;
        if (mutex == null) {
            synchronized (this) {
                mutex = mutexDoNotUseDirectly;
                ) {
                    mutexDoNotUseDirectly = mutex =  Object();
                }
            }
        }
         mutex;
    }
    
    
     * 从RateLimiter中获取一个permit，阻塞直到请求可以获得为止
     * @return 休眠的时间，单位是秒，如果没有被限制则是0.0
      acquire() {
        return acquire(1);
    }
  
    
     * 从RateLimiter中获取指定数量的permits，阻塞直到请求可以获得为止
     double acquire(int permits) {
        long microsToWait = reserve(permits);
        stopwatch.sleepMicrosUninterruptibly(microsToWait);
        return 1.0 * microsToWait / SECONDS.toMicros(1L);
    }
    
    
     * 预定给定数量的permits以备将来使用
     * 直到这些预定数量的permits可以被消费则返回逝去的微秒数
     final long reserve( permits) {
        checkPermits(permits);
        synchronized (mutex()) {
             reserveAndGetWaitLength(permits,stopwatch.readMicros());
        }
    }
    
    void checkPermits( permits) {
        checkArgument(permits > 0,"Requested permits (%s) must be positive"long reserveAndGetWaitLength(int permits,1)"> nowMicros) {
        long momentAvailable = reserveEarliestAvailable(permits,nowMicros);
        return max(momentAvailable - nowMicros,0);
    }
}

class SmoothRateLimiter extends RateLimiter {
    
     The currently stored permits.  storedPermits;

     The maximum number of stored permits.  maxPermits;

    
     * The interval between two unit requests,at our stable rate. E.g.,a stable rate of 5 permits
     * per second has a stable interval of 200ms.
      stableIntervalMicros;
    
    
     * The time when the next request (no matter its size) will be granted. After granting a request,* this is pushed further in the future. Large requests push this further than small requests.
     long nextFreeTicketMicros = 0L;  could be either in the past or future
    
    long reserveEarliestAvailable(int requiredPermits,1)"> nowMicros) {
        resync(nowMicros);
        long returnValue = nextFreeTicketMicros;
        double storedPermitsToSpend = min(requiredPermits,1)">this.storedPermits);        本次可以获取到的permit数量
        double freshPermits = requiredPermits - storedPermitsToSpend;        差值，如果存储的permit大于本次需要的permit数量则此处是0，否则是一个正数
        long waitMicros =
            storedPermitsToWaitTime(.storedPermits,storedPermitsToSpend)
                + (long) (freshPermits * stableIntervalMicros);        计算需要等待的时间（微秒）

        this.nextFreeTicketMicros = LongMath.saturatedAdd(nextFreeTicketMicros,waitMicros);
        this.storedPermits -= storedPermitsToSpend;        减去本次消费的permit数
         returnValue;
    }
    
    void resync( if nextFreeTicket is in the past,resync to now
        if (nowMicros > nextFreeTicketMicros) {        表示当前可以获得permit
            double newPermits = (nowMicros - nextFreeTicketMicros) / coolDownIntervalMicros();        计算这段时间可以生成多少个permit
            storedPermits = min(maxPermits,storedPermits + newPermits);        如果超过maxPermit，则取maxPermit，否则取存储的permit+新生成的permit
            nextFreeTicketMicros = nowMicros;        设置下一次可以获得permit的时间点为当前时间
        }
    }
}

Stopwatch stopwatch = Stopwatch.createStarted();
System.out.println("hahah");
stopwatch.stop();
Duration duration = stopwatch.elapsed();
System.out.println(stopwatch);