第29篇：实战：一起来搞懂限流

star2017 1年前 ⋅ 469 阅读

本文内容

  1. 介绍常见的限流算法
  2. 通过控制最大并发数来进行限流
  3. 通过漏桶算法来进行限流
  4. 通过令牌桶算法来进行限流
  5. 限流工具类RateLimiter

常见的限流的场景

  1. 秒杀活动，数量有限，访问量巨大，为了防止系统宕机，需要做限流处理
  2. 国庆期间，一般的旅游景点人口太多，采用排队方式做限流处理
  3. 医院看病通过发放排队号的方式来做限流处理。

常见的限流算法

  1. 通过控制最大并发数来进行限流
  2. 使用漏桶算法来进行限流
  3. 使用令牌桶算法来进行限流

通过控制最大并发数来进行限流

以秒杀业务为例，10个iphone，100万人抢购，100万人同时发起请求，最终能够抢到的人也就是前面几个人，后面的基本上都没有希望了，那么我们可以通过控制并发数来实现，比如并发数控制在10个，其他超过并发数的请求全部拒绝，提示：秒杀失败，请稍后重试。

并发控制的，通俗解释：一大波人去商场购物，必须经过一个门口，门口有个门卫，兜里面有指定数量的门禁卡，来的人先去门卫那边拿取门禁卡，拿到卡的人才可以刷卡进入商场，拿不到的可以继续等待。进去的人出来之后会把卡归还给门卫，门卫可以把归还来的卡继续发放给其他排队的顾客使用。

JUC中提供了这样的工具类：Semaphore，示例代码：

  1. package com.itsoku.chat29;
  3. import java.util.concurrent.Semaphore;
  4. import java.util.concurrent.TimeUnit;
  6. /**
  7.  * 跟着阿里p7学并发，微信公众号：javacode2018
  8.  */
  9. public class Demo1 {
  11.     static Semaphore semaphore = new Semaphore(5);
  13.     public static void main(String[] args) {
  14.         for (int i = 0; i < 20; i++) {
  15.             new Thread(() -> {
  16.                 boolean flag = false;
  17.                 try {
  18.                     flag = semaphore.tryAcquire(100, TimeUnit.MICROSECONDS);
  19.                     if (flag) {
  20.                         //休眠2秒，模拟下单操作
  21.                         System.out.println(Thread.currentThread() + "，尝试下单中。。。。。");
  22.                         TimeUnit.SECONDS.sleep(2);
  23.                     } else {
  24.                         System.out.println(Thread.currentThread() + "，秒杀失败，请稍微重试！");
  25.                     }
  26.                 } catch (InterruptedException e) {
  27.                     e.printStackTrace();
  28.                 } finally {
  29.                     if (flag) {
  30.                         semaphore.release();
  31.                     }
  32.                 }
  33.             }).start();
  34.         }
  35.     }
  37. }

输出：

  1. Thread[Thread-10,5,main]，尝试下单中。。。。。
  2. Thread[Thread-8,5,main]，尝试下单中。。。。。
  3. Thread[Thread-9,5,main]，尝试下单中。。。。。
  4. Thread[Thread-12,5,main]，尝试下单中。。。。。
  5. Thread[Thread-11,5,main]，尝试下单中。。。。。
  6. Thread[Thread-2,5,main]，秒杀失败，请稍微重试！
  7. Thread[Thread-1,5,main]，秒杀失败，请稍微重试！
  8. Thread[Thread-18,5,main]，秒杀失败，请稍微重试！
  9. Thread[Thread-16,5,main]，秒杀失败，请稍微重试！
  10. Thread[Thread-0,5,main]，秒杀失败，请稍微重试！
  11. Thread[Thread-3,5,main]，秒杀失败，请稍微重试！
  12. Thread[Thread-14,5,main]，秒杀失败，请稍微重试！
  13. Thread[Thread-6,5,main]，秒杀失败，请稍微重试！
  14. Thread[Thread-13,5,main]，秒杀失败，请稍微重试！
  15. Thread[Thread-17,5,main]，秒杀失败，请稍微重试！
  16. Thread[Thread-7,5,main]，秒杀失败，请稍微重试！
  17. Thread[Thread-19,5,main]，秒杀失败，请稍微重试！
  18. Thread[Thread-15,5,main]，秒杀失败，请稍微重试！
  19. Thread[Thread-4,5,main]，秒杀失败，请稍微重试！
  20. Thread[Thread-5,5,main]，秒杀失败，请稍微重试！

关于Semaphore的使用，可以移步：JUC中的Semaphore（信号量）

使用漏桶算法来进行限流

国庆期间比较火爆的景点，人流量巨大，一般入口处会有限流的弯道，让游客进去进行排队，排在前面的人，每隔一段时间会放一拨进入景区。排队人数超过了指定的限制，后面再来的人会被告知今天已经游客量已经达到峰值，会被拒绝排队，让其明天或者以后再来，这种玩法采用漏桶限流的方式。

漏桶算法思路很简单，水（请求）先进入到漏桶里，漏桶以一定的速度出水，当水流入速度过大会直接溢出，可以看出漏桶算法能强行限制数据的传输速率。

漏桶算法示意图：

简陋版的实现，代码如下：

  1. package com.itsoku.chat29;
  3. import java.util.Objects;
  4. import java.util.concurrent.ArrayBlockingQueue;
  5. import java.util.concurrent.BlockingQueue;
  6. import java.util.concurrent.TimeUnit;
  7. import java.util.concurrent.atomic.AtomicInteger;
  8. import java.util.concurrent.locks.LockSupport;
  10. /**
  11.  * 跟着阿里p7学并发，微信公众号：javacode2018
  12.  */
  13. public class Demo2 {
  15.     public static class BucketLimit {
  16.         static AtomicInteger threadNum = new AtomicInteger(1);
  17.         //容量
  18.         private int capcity;
  19.         //流速
  20.         private int flowRate;
  21.         //流速时间单位
  22.         private TimeUnit flowRateUnit;
  23.         private BlockingQueue<Node> queue;
  24.         //漏桶流出的任务时间间隔（纳秒）
  25.         private long flowRateNanosTime;
  27.         public BucketLimit(int capcity, int flowRate, TimeUnit flowRateUnit) {
  28.             this.capcity = capcity;
  29.             this.flowRate = flowRate;
  30.             this.flowRateUnit = flowRateUnit;
  31.             this.bucketThreadWork();
  32.         }
  34.         //漏桶线程
  35.         public void bucketThreadWork() {
  36.             this.queue = new ArrayBlockingQueue<Node>(capcity);
  37.             //漏桶流出的任务时间间隔（纳秒）
  38.             this.flowRateNanosTime = flowRateUnit.toNanos(1) / flowRate;
  39.             Thread thread = new Thread(this::bucketWork);
  40.             thread.setName("漏桶线程-" + threadNum.getAndIncrement());
  41.             thread.start();
  42.         }
  44.         //漏桶线程开始工作
  45.         public void bucketWork() {
  46.             while (true) {
  47.                 Node node = this.queue.poll();
  48.                 if (Objects.nonNull(node)) {
  49.                     //唤醒任务线程
  50.                     LockSupport.unpark(node.thread);
  51.                 }
  52.                 //休眠flowRateNanosTime
  53.                 LockSupport.parkNanos(this.flowRateNanosTime);
  54.             }
  55.         }
  57.         //返回一个漏桶
  58.         public static BucketLimit build(int capcity, int flowRate, TimeUnit flowRateUnit) {
  59.             if (capcity < 0 || flowRate < 0) {
  60.                 throw new IllegalArgumentException("capcity、flowRate必须大于0！");
  61.             }
  62.             return new BucketLimit(capcity, flowRate, flowRateUnit);
  63.         }
  65.         //当前线程加入漏桶，返回false，表示漏桶已满；true：表示被漏桶限流成功，可以继续处理任务
  66.         public boolean acquire() {
  67.             Thread thread = Thread.currentThread();
  68.             Node node = new Node(thread);
  69.             if (this.queue.offer(node)) {
  70.                 LockSupport.park();
  71.                 return true;
  72.             }
  73.             return false;
  74.         }
  76.         //漏桶中存放的元素
  77.         class Node {
  78.             private Thread thread;
  80.             public Node(Thread thread) {
  81.                 this.thread = thread;
  82.             }
  83.         }
  84.     }
  86.     public static void main(String[] args) {
  87.         BucketLimit bucketLimit = BucketLimit.build(10, 60, TimeUnit.MINUTES);
  88.         for (int i = 0; i < 15; i++) {
  89.             new Thread(() -> {
  90.                 boolean acquire = bucketLimit.acquire();
  91.                 System.out.println(System.currentTimeMillis() + " " + acquire);
  92.                 try {
  93.                     TimeUnit.SECONDS.sleep(1);
  94.                 } catch (InterruptedException e) {
  95.                     e.printStackTrace();
  96.                 }
  97.             }).start();
  98.         }
  99.     }
  101. }

代码中BucketLimit.build(10, 60, TimeUnit.MINUTES);创建了一个容量为10，流水为60/分钟的漏桶。

代码中用到的技术有：

  1. BlockingQueue阻塞队列
  2. JUC中的LockSupport工具类，必备技能

使用令牌桶算法来进行限流

令牌桶算法的原理是系统以恒定的速率产生令牌，然后把令牌放到令牌桶中，令牌桶有一个容量，当令牌桶满了的时候，再向其中放令牌，那么多余的令牌会被丢弃；当想要处理一个请求的时候，需要从令牌桶中取出一个令牌，如果此时令牌桶中没有令牌，那么则拒绝该请求。从原理上看，令牌桶算法和漏桶算法是相反的，一个“进水”，一个是“漏水”。这种算法可以应对突发程度的请求，因此比漏桶算法好。

令牌桶算法示意图：

有兴趣的可以自己去实现一个。

限流工具类RateLimiter

Google开源工具包Guava提供了限流工具类RateLimiter，可以非常方便的控制系统每秒吞吐量，示例代码如下：

  1. package com.itsoku.chat29;
  3. import com.google.common.util.concurrent.RateLimiter;
  5. import java.util.Calendar;
  6. import java.util.Date;
  7. import java.util.Objects;
  8. import java.util.concurrent.ArrayBlockingQueue;
  9. import java.util.concurrent.BlockingQueue;
  10. import java.util.concurrent.TimeUnit;
  11. import java.util.concurrent.atomic.AtomicInteger;
  12. import java.util.concurrent.locks.LockSupport;
  14. /**
  15.  * 跟着阿里p7学并发，微信公众号：javacode2018
  16.  */
  17. public class Demo3 {
  19.     public static void main(String[] args) throws InterruptedException {
  20.         RateLimiter rateLimiter = RateLimiter.create(5);//设置QPS为5
  21.         for (int i = 0; i < 10; i++) {
  22.             rateLimiter.acquire();
  23.             System.out.println(System.currentTimeMillis());
  24.         }
  25.         System.out.println("----------");
  26.         //可以随时调整速率，我们将qps调整为10
  27.         rateLimiter.setRate(10);
  28.         for (int i = 0; i < 10; i++) {
  29.             rateLimiter.acquire();
  30.             System.out.println(System.currentTimeMillis());
  31.         }
  32.     }
  33. }

输出：

  1. 1566284028725
  2. 1566284028922
  3. 1566284029121
  4. 1566284029322
  5. 1566284029522
  6. 1566284029721
  7. 1566284029921
  8. 1566284030122
  9. 1566284030322
  10. 1566284030522
  11. ----------
  12. 1566284030722
  13. 1566284030822
  14. 1566284030921
  15. 1566284031022
  16. 1566284031121
  17. 1566284031221
  18. 1566284031321
  19. 1566284031422
  20. 1566284031522
  21. 1566284031622

代码中RateLimiter.create(5)创建QPS为5的限流对象，后面又调用rateLimiter.setRate(10);将速率设为10，输出中分2段，第一段每次输出相隔200毫秒，第二段每次输出相隔100毫秒，可以非常精准的控制系统的QPS。

上面介绍的这些，业务中可能会用到，也可以用来应对面试。

最新资料

更多内容请访问：IT源点

注意：本文归作者所有，未经作者允许，不得转载

#Java高并发教程
阅读全部
相关文章推荐
  • 1
  • 2

全部评论: 0 条

    我有话说: