1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

1. double rate; // leak rate in calls/s

2. double burst; // bucket size in calls

3. long refreshTime; // time for last water refresh

4. double water; // water count at refreshTime

5. refreshWater() {

6. long now = getTimeOfDay();

7. //水隨著時間流逝,不斷流走,最多就流乾到0.

8. water = max(0, water- (now - refreshTime)*rate);

9. refreshTime = now;

10. }

11. bool permissionGranted() {

12. refreshWater();

13. if (water < burst) { // 水桶還沒滿,繼續加1

14. water ++;

15. return true;

16. } else {

17. return false;

18. }

19. }

1
2
3
4
5
6
7

1. final RateLimiter rateLimiter = RateLimiter.create(2.0);

2. void submitTasks(List<Runnable> tasks, Executor executor) {

3. for (Runnable task : tasks) {

4. rateLimiter.acquire(); // may wait

5. executor.execute(task);

6. }

7. }

1
2
3
4
5

1. final RateLimiter rateLimiter = RateLimiter.create(5000.0);

2. void submitPacket(byte[] packet) {

3. rateLimiter.acquire(packet.length);

4. networkService.send(packet);

5. }

1
2
3
4
5

1. if(limiter.tryAcquire()) { //未請求到limiter則立即返回false

2. doSomething();

3. }else{

4. doSomethingElse();

5. }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

1. /**

2. * 建立一個穩定輸出令牌的RateLimiter，保證了平均每秒不超過permitsPerSecond個請求

3. * 當請求到來的速度超過了permitsPerSecond，保證每秒只處理permitsPerSecond個請求

4. * 當這個RateLimiter使用不足(即請求到來速度小於permitsPerSecond)，會囤積最多permitsPerSecond個請求

5. */

6. public static RateLimiter create(double permitsPerSecond);

7. /**

8. * 建立一個穩定輸出令牌的RateLimiter，保證了平均每秒不超過permitsPerSecond個請求

9. * 還包含一個熱身期(warmup period),熱身期內，RateLimiter會平滑的將其釋放令牌的速率加大，直到起達到最大速率

10. * 同樣，如果RateLimiter在熱身期沒有足夠的請求(unused),則起速率會逐漸降低到冷卻狀態

11. *

12. * 設計這個的意圖是為了滿足那種資源提供方需要熱身時間，而不是每次訪問都能提供穩定速率的服務的情況(比如帶快取服務，需要定期重新整理快取的)

13. * 引數warmupPeriod和unit決定了其從冷卻狀態到達最大速率的時間

14. */

15. public static RateLimiter create(double permitsPerSecond, long warmupPeriod, TimeUnit unit);

1
2
3

1. public double acquire();

2. public double acquire(int permits);

1
2
3
4
5
6

1. public boolean tryAcquire();

2. //嘗試獲取一個令牌,立即返回

3. public boolean tryAcquire(int permits);

4. public boolean tryAcquire(long timeout, TimeUnit unit);

5. //嘗試獲取permits個令牌,帶超時時間

6. public boolean tryAcquire(int permits, long timeout, TimeUnit unit);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

1. public double acquire() {

2. return acquire(1);

3. }

4. public double acquire(int permits) {

5. long microsToWait = reserve(permits);

6. stopwatch.sleepMicrosUninterruptibly(microsToWait);

7. return 1.0 * microsToWait / SECONDS.toMicros(1L);

8. }

9. final long reserve(int permits) {

10. checkPermits(permits);

11. synchronized (mutex()) { //應對併發情況需要同步

12. return reserveAndGetWaitLength(permits, stopwatch.readMicros());

13. }

14. }

15. final long reserveAndGetWaitLength(int permits, long nowMicros) {

16. long momentAvailable = reserveEarliestAvailable(permits, nowMicros);

17. return max(momentAvailable - nowMicros, 0);

18. }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22

1. final long reserveEarliestAvailable(int requiredPermits, long nowMicros) {

2. resync(nowMicros); //補充令牌

3. long returnValue = nextFreeTicketMicros;

4. //這次請求消耗的令牌數目

5. double storedPermitsToSpend = min(requiredPermits, this.storedPermits);

6. double freshPermits = requiredPermits - storedPermitsToSpend;

7. long waitMicros = storedPermitsToWaitTime(this.storedPermits, storedPermitsToSpend)

8. + (long) (freshPermits * stableIntervalMicros);

9. this.nextFreeTicketMicros = nextFreeTicketMicros + waitMicros;

10. this.storedPermits -= storedPermitsToSpend;

11. return returnValue;

12. }

13. private void resync(long nowMicros) {

14. // if nextFreeTicket is in the past, resync to now

15. if (nowMicros > nextFreeTicketMicros) {

16. storedPermits = min(maxPermits,

17. storedPermits + (nowMicros - nextFreeTicketMicros) / stableIntervalMicros);

18. nextFreeTicketMicros = nowMicros;

19. }

20. }

1
2
3
4
5
6
7
8
9

1. /**

2. * Translates a specified portion of our currently stored permits which we want to

3. * spend/acquire, into a throttling time. Conceptually, this evaluates the integral

4. * of the underlying function we use, for the range of

5. * [(storedPermits - permitsToTake), storedPermits].

6. *

7. * <p>This always holds: {@code 0 <= permitsToTake <= storedPermits}

8. */

9. abstract long storedPermitsToWaitTime(double storedPermits, double permitsToTake);

1

public static RateLimiter create(double permitsPerSecond)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

1. /**

2. * This implements a "bursty" RateLimiter, where storedPermits are translated to

3. * zero throttling. The maximum number of permits that can be saved (when the RateLimiter is

4. * unused) is defined in terms of time, in this sense: if a RateLimiter is 2qps, and this

5. * time is specified as 10 seconds, we can save up to 2 * 10 = 20 permits.

6. */

7. static final class SmoothBursty extends SmoothRateLimiter {

8. /** The work (permits) of how many seconds can be saved up if this RateLimiter is unused? */

9. final double maxBurstSeconds;

10. SmoothBursty(SleepingStopwatch stopwatch, double maxBurstSeconds) {

11. super(stopwatch);

12. this.maxBurstSeconds = maxBurstSeconds;

13. }

14. void doSetRate(double permitsPerSecond, double stableIntervalMicros) {

15. double oldMaxPermits = this.maxPermits;

16. maxPermits = maxBurstSeconds * permitsPerSecond;

17. System.out.println("maxPermits=" + maxPermits);

18. if (oldMaxPermits == Double.POSITIVE_INFINITY) {

19. // if we don't special-case this, we would get storedPermits == NaN, below

20. storedPermits = maxPermits;

21. } else {

22. storedPermits = (oldMaxPermits == 0.0)

23. ? 0.0 // initial state

24. : storedPermits * maxPermits / oldMaxPermits;

25. }

26. }

27. long storedPermitsToWaitTime(double storedPermits, double permitsToTake) {

28. return 0L;

29. }

30. }

1
2
3
4
5
6
7
8
9
10
11

1. maxPermits=4.0, storedPermits=7.2E-4, stableIntervalMicros=250000.0, nextFreeTicketMicros=1472

2. acquire(1), sleepSecond=0.0

3. maxPermits=4.0, storedPermits=3.012212, stableIntervalMicros=250000.0, nextFreeTicketMicros=1004345

4. acquire(3), sleepSecond=0.0

5. maxPermits=4.0, storedPermits=4.0, stableIntervalMicros=250000.0, nextFreeTicketMicros=2004668

6. acquire(10), sleepSecond=0.0

7. maxPermits=4.0, storedPermits=0.0, stableIntervalMicros=250000.0, nextFreeTicketMicros=3504668

8. acquire(1), sleepSecond=0.499591

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

1. static final class SmoothWarmingUp extends SmoothRateLimiter {

2. private final long warmupPeriodMicros;

3. /**

4. * The slope of the line from the stable interval (when permits == 0), to the cold interval

5. * (when permits == maxPermits)

6. */

7. private double slope;

8. private double halfPermits;

9. SmoothWarmingUp(SleepingStopwatch stopwatch, long warmupPeriod, TimeUnit timeUnit) {

10. super(stopwatch);

11. this.warmupPeriodMicros = timeUnit.toMicros(warmupPeriod);

12. }

13. @Override

14. void doSetRate(double permitsPerSecond, double stableIntervalMicros) {

15. double oldMaxPermits = maxPermits;

16. maxPermits = warmupPeriodMicros / stableIntervalMicros;

17. halfPermits = maxPermits / 2.0;

18. // Stable interval is x, cold is 3x, so on average it's 2x. Double the time -> halve the rate

19. double coldIntervalMicros = stableIntervalMicros * 3.0;

20. slope = (coldIntervalMicros - stableIntervalMicros) / halfPermits;

21. if (oldMaxPermits == Double.POSITIVE_INFINITY) {

22. // if we don't special-case this, we would get storedPermits == NaN, below

23. storedPermits = 0.0;

24. } else {

25. storedPermits = (oldMaxPermits == 0.0)

26. ? maxPermits // initial state is cold

27. : storedPermits * maxPermits / oldMaxPermits;

28. }

29. }

30. @Override

31. long storedPermitsToWaitTime(double storedPermits, double permitsToTake) {

32. double availablePermitsAboveHalf = storedPermits - halfPermits;

33. long micros = 0;

34. // measuring the integral on the right part of the function (the climbing line)

35. if (availablePermitsAboveHalf > 0.0) {

36. double permitsAboveHalfToTake = min(availablePermitsAboveHalf, permitsToTake);

37. micros = (long) (permitsAboveHalfToTake * (permitsToTime(availablePermitsAboveHalf)

38. + permitsToTime(availablePermitsAboveHalf - permitsAboveHalfToTake)) / 2.0);

39. permitsToTake -= permitsAboveHalfToTake;

40. }

41. // measuring the integral on the left part of the function (the horizontal line)

42. micros += (stableIntervalMicros * permitsToTake);

43. return micros;

44. }

45. private double permitsToTime(double permits) {

46. return stableIntervalMicros + permits * slope;

47. }

48. }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

1. * ^ throttling

2. * |

3. * 3*stable + /

4. * interval | /.

5. * (cold) | / .

6. * | / . <-- "warmup period" is the area of the trapezoid between

7. * 2*stable + / . halfPermits and maxPermits

8. * interval | / .

9. * | / .

10. * | / .

11. * stable +----------/ WARM . }

12. * interval | . UP . } <-- this rectangle (from 0 to maxPermits, and

13. * | . PERIOD. } height == stableInterval) defines the cooldown period,

14. * | . . } and we want cooldownPeriod == warmupPeriod

15. * |---------------------------------> storedPermits

16. * (halfPermits) (maxPermits)

17. *

1
2
3
4
5
6
7
8
9
10
11
12
13
14

1. warmupPeriodMicros=2000000

2. stableIntervalMicros=250000.0, maxPermits=8.0, halfPermits=4.0, coldIntervalMicros=750000.0, slope=125000.0, storedPermits=8.0

3. maxPermits=8.0, storedPermits=8.0, stableIntervalMicros=250000.0, nextFreeTicketMicros=1524

4. acquire(1), sleepSecond=0.0

5. maxPermits=8.0, storedPermits=8.0, stableIntervalMicros=250000.0, nextFreeTicketMicros=1001946

6. acquire(3), sleepSecond=0.0

7. maxPermits=8.0, storedPermits=5.0, stableIntervalMicros=250000.0, nextFreeTicketMicros=2689446

8. acquire(10), sleepSecond=0.687186

9. maxPermits=8.0, storedPermits=0.0, stableIntervalMicros=250000.0, nextFreeTicketMicros=5251946

10. acquire(1), sleepSecond=1.559174