2017-09-23

Google Guava 缓存实现接口的限流

项目背景

最近项目中需要进行接口保护，防止高并发的情况把系统搞崩，因此需要对一个查询接口进行限流，主要的目的就是限制单位时间内请求此查询的次数，例如 1000 次，来保护接口。
参考了开涛的博客聊聊高并发系统限流特技，学习了其中利用 Google Guava 缓存实现限流的技巧，在网上也查到了很多关于 Google Guava 缓存的博客，学到了好多，推荐一个博客文章：http://ifeve.com/google-guava-cachesexplained/, 关于 Google Guava 缓存的更多细节或者技术，这篇文章讲的很详细；
这里我们并不是用缓存来优化查询，而是利用缓存，存储一个计数器，然后用这个计数器来实现限流。

效果实验

static LoadingCache<Long, AtomicLong> count = CacheBuilder.newBuilder().expireAfterWrite(1, TimeUnit.SECONDS).build(new CacheLoader<Long, AtomicLong>() {
            @Override
        public AtomicLong load(Long o) throws Exception {
            //System.out.println("Load call!");
            return new AtomicLong(0L);
        }
    });

上面，我们通过 CacheBuilder 来新建一个 LoadingCache 缓存对象 count，然后设置其有效时间为 1 秒，即每 1 秒钟刷新一次；缓存中，key 为一个 long 型的时间戳类型，value 是一个计数器，使用原子性的 AtomicLong 保证自增和自减操作的原子性，每次查询缓存时如果不能命中，即查询的时间戳不在缓存中，则重新加载缓存，执行 load 将当前的时间戳的计数值初始化为 0。这样对于每一秒的时间戳，能计算这一秒内执行的次数，从而达到限流的目的；
这是要执行的一个 getCounter 方法：

public class Counter {
    static int counter = 0;
    public static int getCounter() throws Exception{
        return counter++;
    }
}

现在我们创建多个线程来执行这个方法：

public class Test {

    public static void main(String args[]) throws Exception
    {
        for(int i = 0;i<100;i++)
        {
            new Thread(){
                @Override
                public void run() {
                    try {
                        System.out.println(Counter.getCounter());
                    }
                    catch (Exception e)
                    {
                        e.printStackTrace();
                    }
                }
            }.start();
        }

    }
}

这样执行的话，执行结果很简单，就是很快地执行这个 for 循环，迅速打印 0 到 99 折 100 个数，不再贴出。
这里的 for 循环执行 100 个进程时间是很快的，那么现在我们要限制每秒只能有 10 个线程来执行 getCounter() 方法，该怎么办呢，上面讲的限流方法就派上用场了：

public class Counter {
    static LoadingCache<Long, AtomicLong> count = CacheBuilder.newBuilder().expireAfterWrite(1, TimeUnit.SECONDS).build(new CacheLoader<Long, AtomicLong>() {
            @Override
        public AtomicLong load(Long o) throws Exception {
                System.out.println("Load call!");
            return new AtomicLong(0L);
        }
    });
    static long limits = 10;
    static int counter = 0;
    public static synchronized int getCounter() throws Exception{
        while (true)
        {
            //获取当前的时间戳作为key
            Long currentSeconds = System.currentTimeMillis() / 1000;
            if (count.get(currentSeconds).getAndIncrement() > limits) {
                continue;
            }
            return counter++;
        }
    }
}

这样一来，就可以限制每秒的执行数了。对于每个线程，获取当前时间戳，如果当前时间 (当前这 1 秒) 内有超过 10 个线程正在执行，那么这个进程一直在这里循环，直到下一秒，或者更靠后的时间，重新加载，执行 load，将新的时间戳的计数值重新为 0。
执行结果：

每秒执行 11 个（因为从 0 开始），每一秒之后，load 方法会执行一次；

为了更加直观，我们可以让每个for循环sleep一段时间：

public class Test {

    public static void main(String args[]) throws Exception
    {
        for(int i = 0;i<100;i++)
        {
            new Thread(){
                @Override
                public void run() {
                    try {
                        System.out.println(Counter.getCounter());
                    }
                    catch (Exception e)
                    {
                        e.printStackTrace();
                    }
                }
            }.start();
            Thread.sleep(100);
        }

    }
}