限流RateLimiter

限流算法, Guava RateLimiter使用、设计及其实现.

1、限流

常见并发系统的限流：

  • 限制并发总数(数据库连接池、线程池)
  • 限制瞬时并发数(nginx的limit_conn)
  • 限制时间窗口内的平均塑料
  • 限制远程接口调用速率
  • 限制MQ的消费速率

限流算法：

  • 令牌桶: 存放固定容量令牌的桶，按照固定速率往桶里面添加令牌
  • 漏桶:　用于流量整形和流量控制，固定容量的漏桶，按照常量固定速率流出
  • 计数器

2、SimpleRateLimiter

一个简单的RateLimiter

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
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;


public class SimpleRateLimiter {
    private int qps;
    private LoadingCache<Long, AtomicInteger> cache = CacheBuilder.newBuilder()
            .expireAfterWrite(2, TimeUnit.SECONDS)
            .recordStats()
            .build(new CacheLoader<Long, AtomicInteger>() {
                @Override
                public AtomicInteger load(Long key) throws Exception {
                    return new AtomicInteger(0);
                }
            });

    public SimpleRateLimiter(int qps) {
        this.qps = qps;
    }

    public long getSize() {
        return cache.size();
    }


    public boolean acquire() {
        try {
            Long cur = System.currentTimeMillis() / 1000;
            if (cache.get(cur).getAndIncrement() < qps) {
                return true;
            }

        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        return false;

    }


    public static void main(String[] ar) {
        ExecutorService exe = Executors.newFixedThreadPool(10);
        final SimpleRateLimiter limiter = new SimpleRateLimiter(1);


        for (int i = 0; i < 10; i++) {
            final int j = i;
            exe.submit(new Thread(new Runnable() {
                @Override
                public void run() {
                    while (true) {
                        if (limiter.acquire()) {
                            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
                            System.out.printf("Thread-%d [%s] : do its work, cache size=%d!\n", j, format.format(new Date()), limiter.getSize());
                        }
                    }
                }
            }));
        }

    }

}

3、Guava RateLimiter