限流(二)接口限流

如果某个接口可能出现突发情况，比如“秒杀”活动，那么很有可能因为突然爆发的访问量造成系统奔溃，我们需要最这样的接口进行限流。

在上一篇“限流算法”中，我们简单提到了两种限流方式：

1）（令牌桶、漏桶算法）限速率，例如：每 5r/1s = 1r/200ms 即一个请求以200毫秒的速率来执行；

2）（计数器方式）限制总数、或者单位时间内的总数，例如：设定总并发数的阀值，单位时间总并发数的阀值。

一、限制总并发数

我们可以采用java提供的atomicLong类来实现

atomicLong在java.util.concurrent.atomic包下，它直接继承于number类，它是线程安全的。

atomicLong可以参考：http://www.cnblogs.com/lay2017/p/9066719.html

CountDownLatch可以参考：http://www.cnblogs.com/lay2017/p/9067756.html

我们将使用它来计数

public class AtomicDemo {
    // 计数
    public static AtomicLong atomicLong = new AtomicLong(0L);
    // 最大请求数量
    static int limit = 10;
    // 请求数量
    static int reqAmonut = 15;
    
    public static void main(String[] args) throws InterruptedException {
        // 多线程并发模拟
        final CountDownLatch latch = new CountDownLatch(1);
        for (int i = 1; i <= reqAmonut; i++) {
            final int t = i;
            new Thread(new Runnable() {
                
                public void run() {
                    try {
                        latch.await();
                        // 计数器加1，并判断最大请求数量
                        if (atomicLong.getAndIncrement() > limit) {
                            System.out.println(t + "线程：限流了");
                            return;
                        } 
                        System.out.println(t + "线程：业务处理");
                        // 休眠1秒钟，模拟业务处理
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        // 计数器减1
                        atomicLong.decrementAndGet();
                    }
                }
            }).start();
        }
        latch.countDown();
    }
}
 

二、限制单位时间的总并发数

下面用谷歌的Guava依赖中的Cache（线程安全）来完成单位时间的并发数限制，

Guava需要引入依赖：

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>21.0</version>
</dependency>

具体逻辑如下：

1）根据当前的的时间戳（秒）做key，请求计数的值做value；

2）每个请求都通过时间戳来获取计数值，并判断是否超过限制。（即，1秒内的请求数量是否超过阀值）

代码如下：

public class AtomicDemo2 {
    // 计数
    public static AtomicLong atomicLong = new AtomicLong(0L);
    // 最大请求数量
    static int limit = 10;
    // 请求数量
    static int reqAmonut = 15;
    
    public static void main(String[] args) throws InterruptedException {
        // Guava的Cache来存储计数器

        final LoadingCache<Long, AtomicLong> counter = CacheBuilder.newBuilder()
　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　.expireAfterWrite(1, TimeUnit.SECONDS)
　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　　.build(new CacheLoader<Long, AtomicLong>(){
            　　　　　　　　　　　　　　　　　　　　　　　　　　　　@Override
           　　　　　　　　 　　　　　　　　　　　　　　　　　　　　public AtomicLong load(Long key) throws Exception {
                 　　　　　　　　　　　　　　　　　　　　　　　　　　　　return new AtomicLong(0L);
            　　　　　　　　　　　　　　　　　　　　　　　　　　　　}
        　　　　　　　　　　　　　　　　　　　　　　　　　　　　　});
        
        // 多线程并发模拟
        final CountDownLatch latch = new CountDownLatch(1);
        for (int i = 1; i <= reqAmonut; i++) {
            final int t = i;
            new Thread(new Runnable() {
                
                public void run() {
                    try {
                        latch.await();
                        long currentSeconds = System.currentTimeMillis()/1000;
                        // 从缓存中取值，并计数器+1
                        if (counter.get(currentSeconds).getAndIncrement() > limit) {
                            System.out.println(t + "线程：限流了");
                            return;
                        } 
                        System.out.println(t + "线程：业务处理");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (ExecutionException e) {
                        e.printStackTrace();  
                    }
                }
            }).start();
        }
        latch.countDown();
    }
}
 

三、限制接口的速率

以上两种以较为简单的计数器方式实现了限流，但是他们都只是限制了总数。也就是说，它们允许瞬间爆发的请求达到最大值，这有可能导致一些问题。

下面我们将使用Guava的 RateLimiter提供的令牌桶算法来实现限制速率，例如：1r/200ms

同样需要引入依赖：

<dependency>
    <groupId>com.google.guava</groupId>
    <artifactId>guava</artifactId>
    <version>21.0</version>
</dependency>

示例代码：

public class GuavaDemo {
    // 每秒钟5个令牌
    static RateLimiter limiter = RateLimiter.create(5);
    
    public static void main(String[] args) throws InterruptedException {
        final RateLimiter limiter2 = RateLimiter.create(5);
        for (int i = 0; i < 20; i++) {
            System.out.println(i + "-" + limiter2.acquire());
        }
    }
}
  

说明：

1）RateLimiter.create(5)表示创建一个容量为5的令牌桶，并且每秒钟新增5个令牌，也就是每200毫秒新增1个令牌；

2）limiter2.acquire() 表示消费一个令牌，如果桶里面没有足够的令牌那么就进入等待。

输出：

0.0
0.197729
0.192975
...

平均 1r/200ms的速率处理请求

RateLimiter允许突发超额，例如：

public class GuavaDemo {
    // 每秒钟5个令牌
    static RateLimiter limiter = RateLimiter.create(5);
    
    public static void main(String[] args) throws InterruptedException {
        final RateLimiter limiter2 = RateLimiter.create(5);
        System.out.println(limiter2.acquire(10));
        System.out.println(limiter2.acquire());
        System.out.println(limiter2.acquire());
        System.out.println(limiter2.acquire());
        System.out.println(limiter2.acquire());
        System.out.println(limiter2.acquire());
        System.out.println(limiter2.acquire());
    }
}

输出：

0.0
1.997777
0.194835
0.198466
0.195192
0.197448
0.196706

我们看到：

limiter2.acquire(10)

超额消费了10个令牌，而下一个消费需要等待超额消费的时间，所以等待了近2秒钟的时间，而后又开始匀速处理请求

由于上面的方式允许突发，很多人可能担心这种突发对于系统来说如果扛不住可能就造成崩溃。那针对这种情况，大家希望能够从慢速到匀速地平滑过渡。Guava当然也提供了这样的实现：

public class GuavaDemo {
    // 每秒钟5个令牌
    static RateLimiter limiter = RateLimiter.create(5);
    
    public static void main(String[] args) throws InterruptedException {
        final RateLimiter limiter2 = RateLimiter.create(5, 1, TimeUnit.SECONDS);
        System.out.println(limiter2.acquire());
        System.out.println(limiter2.acquire());
        System.out.println(limiter2.acquire());
        System.out.println(limiter2.acquire());
        System.out.println(limiter2.acquire());
        System.out.println(limiter2.acquire());
        System.out.println(limiter2.acquire());
    }
}

输出：

0.0
0.51798
0.353722
0.216954
0.195776
0.194903
0.194547

我们看到，速率从0.5慢慢地趋于0.2，平滑地过渡到了匀速状态。

RateLimter 还提供了tryAcquire()方法来判断是否有够的令牌，并即时返回结果，如：

public class GuavaDemo {
    public static void main(String[] args) throws InterruptedException {
        final RateLimiter limiter = RateLimiter.create(5, 1, TimeUnit.SECONDS);
        for (int i = 0; i < 10; i++) {
            if (limiter.tryAcquire()) {
                System.out.println("处理业务");
            }else{
                System.out.println("限流了");
            }
        }
    }
}

输出：

处理业务
限流了
限流了
限流了
限流了
限流了
限流了
限流了
限流了
限流了

以上，就是单实例系统的应用级接口限流方式。

参考：

http://jinnianshilongnian.iteye.com/blog/2305117