Java多线程 5.栅栏

Posted 2020-12-03 kepus

1.Java多线程-认识Java线程

2.Java多线程-线程安全

3.Java多线程-线程协作

4.Java多线程-线程池

5.Java多线程-栅栏

6.Java多线程-Fork/Join

 

---

5.1 ReadMe

　　此文线程和任务可以理解为一个意思；

　　Java中一般通过CountDownLantch和CyclicBarrier来解决线程（任务）之间依赖的问题，栅栏特指CyclicBarrier类，因为CountDownLatch可以实现类似功能，所以在此放到一块讲解；

　　在任务A依赖任务B的这种场景可以使用Object的wait和notify来实现，但是如果任务A依赖任务B、C、D多个任务的场景，使用Object的wait和notify就难以实现，例如运动会10个人长跑（看作10个长跑任务），公布总排名这个任务就依赖至少9个长跑任务结束，这种场景适合使用CountDownLatch；

　　结合实际开发过程，更多的场景是A、B、C多个任务同时执行，但是A、B、C任务在执行过程中的某一个点相互依赖，例如一个需求分为前段开发和后端开发，前后端开始联调的时间点就是2个任务相互依赖的点，这种场景适合使用CyclicBarrier；

5.2 CountDownLatch

　　Latch是门闩的意思，要打开门需要打开门上的所有门闩，CountDownLatch可以理解为一个有多个门闩的门，每个门闩需要一个的线程打开；

　　代码实现举例：

 1 import java.security.SecureRandom;
 2 import java.util.concurrent.CountDownLatch;
 3 import java.util.concurrent.ExecutorService;
 4 import java.util.concurrent.Executors;
 5 import java.util.concurrent.TimeUnit;
 6 
 7 /**
 8  * CountDownLatch实例
 9  */
10 public class DemoCountDownLatch {
11 
12     //参赛人数 / 门闩个数
13     private static final int COUNT_RUNNER = 3;
14 
15     public static void main(String[] args) throws Exception {
16 
17         // 3人参加比赛 / 创建一个有3个门闩的门
18         CountDownLatch countDownLatch = new CountDownLatch(COUNT_RUNNER);
19 
20         // 创建3人跑道 / 装上门
21         ExecutorService executorService = Executors.newFixedThreadPool(COUNT_RUNNER);
22 
23         for (int i = 0; i < 3; i++) {
24             executorService.submit(new Runner("runner" + i, countDownLatch));
25         }
26 
27         // 等待3人跑到终点 / 把3个门闩都锁上
28         countDownLatch.await();
29 
30         // 公布3人成绩 / 打开门了
31         System.out.println("all runner game over.");
32     }
33 
34     static class Runner implements Runnable {
35 
36         private String name;
37 
38         private CountDownLatch countDownLatch;
39 
40         public Runner(String name, CountDownLatch countDownLatch) {
41             this.name = name;
42             this.countDownLatch = countDownLatch;
43         }
44 
45         @Override
46         public void run() {
47             try {
48                 SecureRandom secureRandom = SecureRandom.getInstanceStrong();
49                 int runTime = Math.abs(secureRandom.nextInt()) % 10;
50                 TimeUnit.SECONDS.sleep(runTime);
51                 System.out.println(this.name + " game over cost " + runTime + " second.");
52             } catch (Exception e) {
53                 e.printStackTrace();
54             }
55 
56             // 跑到终点 / 打开门闩
57             this.countDownLatch.countDown();
58         }
59     }
60 }

View Code

　 说明：

1. 一个CountDownLatch对象后只能使用一次，也就是说不能工作同一个CountDownLatch对象来重复控制线程的依赖问题;

2.上面的例子中如果有长跑运动员中途放弃比赛，是否永远不能公布总的比赛成绩？ CountDownLatch的await可以有入参（timeout, TimeUnit）表示最长等待时间;

5.3 CyclicBarrier

 CyclicBarrier是循环栅栏的的意思，循环表示同一个CyclicBarrier可以重复使用（区别于CountDownLatch），Barrier栅栏可以理解为线程相互依赖的那个点（例如前后端联调时间点），各个线程在那个点相互等待，等所有线程到达点后才继续执行；

代码实现举例：

 1 import java.time.Instant;
 2 import java.util.concurrent.*;
 3 
 4 /**
 5  * 前端开发倩倩和后端开发厚厚开发一个需求
 6  * 两人先独自开发需求，等都开发完再一块联调功能
 7  */
 8 public class DemoCyclicBarrier
 9 {
10     public static void main(String[] args) throws InterruptedException
11     {
12 
13         // 创建栅栏，参数一为相互依赖的任务数；参数二为各任务到达依赖点后先执行的任务，等任务执行结束相互依赖的任务继续执行
14         CyclicBarrier cyclicBarrier = new CyclicBarrier(2, () -> {
15             System.out.println("准备开始联调吧...");
16             lastSecond(3L);
17         });
18 
19         // 创建线程池执行2个任务
20         ExecutorService executorService = Executors.newFixedThreadPool(2);
21         executorService.execute(new Coder(10L, "后端", cyclicBarrier));
22         executorService.execute(new Coder(3L, "前端", cyclicBarrier));
23     }
24 
25     /**
26      * 线程持续second秒
27      */
28     private static void lastSecond(long second)
29     {
30         Instant instant = Instant.now();
31         while (Instant.now().minusSeconds(second).isBefore(instant))
32         {
33         }
34     }
35 
36     static class Coder implements Runnable
37     {
38         // 完成工作需要的时间
39         private long workTime;
40 
41         private String name;
42 
43         private CyclicBarrier cyclicBarrier;
44 
45         public Coder(long workTime, String name, CyclicBarrier cyclicBarrier)
46         {
47             this.workTime = workTime;
48             this.name = name;
49             this.cyclicBarrier = cyclicBarrier;
50         }
51 
52         @Override
53         public void run()
54         {
55             try
56             {
57                 System.out.println(this.name + " are coding...");
58                 lastSecond(this.workTime);
59                 System.out.println(this.name + " code end wait debugging..");
60                 // 完成工作/到达依赖的点/我这边可以开始联调了
61                 this.cyclicBarrier.await();
62                 System.out.println("we are debugging..");
63 
64             }
65             catch (InterruptedException e)
66             {
67                 // 当前线程被中断
68                 e.printStackTrace();
69             }
70             catch (BrokenBarrierException e)
71             {
72                 // 1.其他线程中断；2.其他线程await方法超时；3.cyclicBarrier重置
73                 e.printStackTrace();
74             }
75             // catch (TimeoutException e)
76             // {
77             // //当前线程的await方法超时(await方法设置超时参数)
78             // e.printStackTrace();
79             // }
80         }
81     }
82 }

View Code

说明：

1.CyclicBarrier作为循环栅栏，同一个对象可以循环使用；

2.上面例子中前端开发人员很短时间开发结束，通过await()一直在等待后端开发结束，可以通过await(timeout, TimeUnit)来设置最长等待时间；

3. 可以通过CyclicBarrier的getNumberWaiting()查看到达依赖点的任务；

4.CyclicBarrier构造方法的第二个参数指定的任务A，在其他相互依赖的任务到达依赖点后，任务A优先执行，并且是执行结束，其他任务才继续执行；

5.4 CyclicBarrier&CountDownLatch举例

 1 import java.time.Instant;
 2 import java.util.concurrent.*;
 3 
 4 public class Demo_CyclicBarrier_CountDownLatch
 5 {
 6     private static final int COUNT_WORKER = 2;
 7 
 8     public static void main(String[] args) throws InterruptedException
 9     {
10         CountDownLatch countDownLatch = new CountDownLatch(COUNT_WORKER);
11         CyclicBarrier cyclicBarrier = new CyclicBarrier(COUNT_WORKER, () -> {
12             System.out.println("准备开始联调吧...");
13             lastSecond(3L);
14         });
15 
16         ExecutorService executorService = Executors.newFixedThreadPool(2);
17         executorService.execute(new Coder(10L, "后端", countDownLatch, cyclicBarrier));
18         executorService.execute(new Coder(3L, "前端", countDownLatch, cyclicBarrier));
19 
20         countDownLatch.await();
21         System.out.println("开发联调结束，需求交付...");
22     }
23 
24     /**
25      * 线程持续second秒
26      */
27     private static void lastSecond(long second)
28     {
29         Instant instant = Instant.now();
30         while (Instant.now().minusSeconds(second).isBefore(instant))
31         {
32         }
33     }
34 
35     static class Coder implements Runnable
36     {
37         // 开发联调时间
38         private long workTime;
39 
40         private String name;
41 
42         private CountDownLatch countDownLatch;
43 
44         private CyclicBarrier cyclicBarrier;
45 
46         public Coder(long workTime, String name, CountDownLatch countDownLatch, CyclicBarrier cyclicBarrier)
47         {
48             this.workTime = workTime;
49             this.name = name;
50             this.countDownLatch = countDownLatch;
51             this.cyclicBarrier = cyclicBarrier;
52         }
53 
54         @Override
55         public void run()
56         {
57             try
58             {
59                 System.out.println(this.name + " are coding...");
60                 lastSecond(this.workTime);
61                 System.out.println(this.name + " code end wait debugging..");
62 
63                 this.cyclicBarrier.await();
64 
65                 System.out.println(this.name + " are debugging..");
66                 lastSecond(this.workTime);
67 
68                 System.out.println(this.name + " debug end..");
69                 this.countDownLatch.countDown();
70             }
71             catch (Exception e)
72             {
73                 e.printStackTrace();
74             }
75         }
76     }
77 }

View Code