java并发系列 – 第29天:高并发中常见的限流方式

  • 2019 年 10 月 3 日
  • 筆記

这是java高并发系列第29篇。

环境:jdk1.8。

本文内容

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

常见的限流的场景

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

常见的限流算法

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

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

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

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

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

package com.itsoku.chat29;    import java.util.concurrent.Semaphore;  import java.util.concurrent.TimeUnit;    /**   * 跟着阿里p7学并发,微信公众号:javacode2018   */  public class Demo1 {        static Semaphore semaphore = new Semaphore(5);        public static void main(String[] args) {          for (int i = 0; i < 20; i++) {              new Thread(() -> {                  boolean flag = false;                  try {                      flag = semaphore.tryAcquire(100, TimeUnit.MICROSECONDS);                      if (flag) {                          //休眠2秒,模拟下单操作                          System.out.println(Thread.currentThread() + ",尝试下单中。。。。。");                          TimeUnit.SECONDS.sleep(2);                      } else {                          System.out.println(Thread.currentThread() + ",秒杀失败,请稍微重试!");                      }                  } catch (InterruptedException e) {                      e.printStackTrace();                  } finally {                      if (flag) {                          semaphore.release();                      }                  }              }).start();          }      }    }

输出:

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

关于Semaphore的使用,可以移步:JUC中的Semaphore(信号量)

使用漏桶算法来进行限流

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

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

漏桶算法示意图:

简陋版的实现,代码如下:

package com.itsoku.chat29;    import java.util.Objects;  import java.util.concurrent.ArrayBlockingQueue;  import java.util.concurrent.BlockingQueue;  import java.util.concurrent.TimeUnit;  import java.util.concurrent.atomic.AtomicInteger;  import java.util.concurrent.locks.LockSupport;    /**   * 跟着阿里p7学并发,微信公众号:javacode2018   */  public class Demo2 {        public static class BucketLimit {          static AtomicInteger threadNum = new AtomicInteger(1);          //容量          private int capcity;          //流速          private int flowRate;          //流速时间单位          private TimeUnit flowRateUnit;          private BlockingQueue<Node> queue;          //漏桶流出的任务时间间隔(纳秒)          private long flowRateNanosTime;            public BucketLimit(int capcity, int flowRate, TimeUnit flowRateUnit) {              this.capcity = capcity;              this.flowRate = flowRate;              this.flowRateUnit = flowRateUnit;              this.bucketThreadWork();          }            //漏桶线程          public void bucketThreadWork() {              this.queue = new ArrayBlockingQueue<Node>(capcity);              //漏桶流出的任务时间间隔(纳秒)              this.flowRateNanosTime = flowRateUnit.toNanos(1) / flowRate;              Thread thread = new Thread(this::bucketWork);              thread.setName("漏桶线程-" + threadNum.getAndIncrement());              thread.start();          }            //漏桶线程开始工作          public void bucketWork() {              while (true) {                  Node node = this.queue.poll();                  if (Objects.nonNull(node)) {                      //唤醒任务线程                      LockSupport.unpark(node.thread);                  }                  //休眠flowRateNanosTime                  LockSupport.parkNanos(this.flowRateNanosTime);              }          }            //返回一个漏桶          public static BucketLimit build(int capcity, int flowRate, TimeUnit flowRateUnit) {              if (capcity < 0 || flowRate < 0) {                  throw new IllegalArgumentException("capcity、flowRate必须大于0!");              }              return new BucketLimit(capcity, flowRate, flowRateUnit);          }            //当前线程加入漏桶,返回false,表示漏桶已满;true:表示被漏桶限流成功,可以继续处理任务          public boolean acquire() {              Thread thread = Thread.currentThread();              Node node = new Node(thread);              if (this.queue.offer(node)) {                  LockSupport.park();                  return true;              }              return false;          }            //漏桶中存放的元素          class Node {              private Thread thread;                public Node(Thread thread) {                  this.thread = thread;              }          }      }        public static void main(String[] args) {          BucketLimit bucketLimit = BucketLimit.build(10, 60, TimeUnit.MINUTES);          for (int i = 0; i < 15; i++) {              new Thread(() -> {                  boolean acquire = bucketLimit.acquire();                  System.out.println(System.currentTimeMillis() + " " + acquire);                  try {                      TimeUnit.SECONDS.sleep(1);                  } catch (InterruptedException e) {                      e.printStackTrace();                  }              }).start();          }      }    }

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

代码中用到的技术有:

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

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

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

令牌桶算法示意图:

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

限流工具类RateLimiter

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

package com.itsoku.chat29;    import com.google.common.util.concurrent.RateLimiter;    import java.util.Calendar;  import java.util.Date;  import java.util.Objects;  import java.util.concurrent.ArrayBlockingQueue;  import java.util.concurrent.BlockingQueue;  import java.util.concurrent.TimeUnit;  import java.util.concurrent.atomic.AtomicInteger;  import java.util.concurrent.locks.LockSupport;    /**   * 跟着阿里p7学并发,微信公众号:javacode2018   */  public class Demo3 {        public static void main(String[] args) throws InterruptedException {          RateLimiter rateLimiter = RateLimiter.create(5);//设置QPS为5          for (int i = 0; i < 10; i++) {              rateLimiter.acquire();              System.out.println(System.currentTimeMillis());          }          System.out.println("----------");          //可以随时调整速率,我们将qps调整为10          rateLimiter.setRate(10);          for (int i = 0; i < 10; i++) {              rateLimiter.acquire();              System.out.println(System.currentTimeMillis());          }      }  }

输出:

1566284028725  1566284028922  1566284029121  1566284029322  1566284029522  1566284029721  1566284029921  1566284030122  1566284030322  1566284030522  ----------  1566284030722  1566284030822  1566284030921  1566284031022  1566284031121  1566284031221  1566284031321  1566284031422  1566284031522  1566284031622

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

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

java高并发系列目录

  1. 第1天:必须知道的几个概念
  2. 第2天:并发级别
  3. 第3天:有关并行的两个重要定律
  4. 第4天:JMM相关的一些概念
  5. 第5天:深入理解进程和线程
  6. 第6天:线程的基本操作
  7. 第7天:volatile与Java内存模型
  8. 第8天:线程组
  9. 第9天:用户线程和守护线程
  10. 第10天:线程安全和synchronized关键字
  11. 第11天:线程中断的几种方式
  12. 第12天JUC:ReentrantLock重入锁
  13. 第13天:JUC中的Condition对象
  14. 第14天:JUC中的LockSupport工具类,必备技能
  15. 第15天:JUC中的Semaphore(信号量)
  16. 第16天:JUC中等待多线程完成的工具类CountDownLatch,必备技能
  17. 第17天:JUC中的循环栅栏CyclicBarrier的6种使用场景
  18. 第18天:JAVA线程池,这一篇就够了
  19. 第19天:JUC中的Executor框架详解1
  20. 第20天:JUC中的Executor框架详解2
  21. 第21天:java中的CAS,你需要知道的东西
  22. 第22天:JUC底层工具类Unsafe,高手必须要了解
  23. 第23天:JUC中原子类,一篇就够了
  24. 第24天:ThreadLocal、InheritableThreadLocal(通俗易懂)
  25. 第25天:掌握JUC中的阻塞队列
  26. 第26篇:学会使用JUC中常见的集合,常看看!
  27. 第27天:实战篇,接口性能提升几倍原来这么简单
  28. 第28天:实战篇,微服务日志的伤痛,一并帮你解决掉

java高并发系列连载中,总计估计会有四五十篇文章。

阿里p7一起学并发,公众号:路人甲java,每天获取最新文章!