Java并发编程之-了解CyclicBarrier

Posted 2021-02-14 yuxiaoming

起因：一道面试题，关于concurrent包下的CyclicBarrier并发工具类，提出的一个情景题，因为不了解，所以花了一天的时间去学习并以此记录而诞生的这篇博客。

 

题目：假设有5个运动员（线程），让他们就绪在同一位置开始比赛跑步，当裁判（主线程）发出枪响号令。5个运动员（5个线程执行）开始奔跑，当5个线程跑到终点后。裁判（主线程执行）宣布比赛结果。

 

知识点：CyclicBarrier，CountDownLatch这两个位于JDK concurren包下的两个类。今天只实现CyclicBarrier这个类

原理：        CyclicBarrier大致是可循环利用的屏障，顾名思义，这个名字也将这个类的特点给明确地表示出来了。首先，便是可重复利用，说明该类创建的对象可以复用；其次，屏障则体现了该类的原理：每个线程执行时，都会碰到一个屏障，直到所有线程执行结束，然后屏障便会打开，使所有线程继续往下执行。

        这里介绍CyclicBarrier的两个构造函数：CyclicBarrier(int parties)和CyclicBarrier(int parties, Runnable barrierAction) :前者只需要声明需要拦截的线程数即可，而后者还需要定义一个等待所有线程到达屏障优先执行的Runnable对象。

        实现原理：在CyclicBarrier的内部定义了一个Lock对象，每当一个线程调用await方法时，将拦截的线程数减1，然后判断剩余拦截数是否为初始值parties，如果不是，进入Lock对象的条件队列等待。如果是，执行barrierAction对象的Runnable方法，然后将锁的条件队列中的所有线程放入锁等待队列中，这些线程会依次的获取锁、释放锁。

难点：

　　1 ：5个子线程执行运行完毕后，主线程执行其他任务方法，期间不能死亡。

　　2：个人尝试使用线程池，来管理线程的执行。发现使用线程池子线程执行任务时，主线程会随时挂掉，这里我使用的睡眠sleep，但是我的“小伙伴“说这样不太好，这个问题暂时遗留记录

 

线程池版实现：

 1 import java.util.concurrent.*;
 2 
 3 /**
 4 * Java并发编程之，了解CyclicBarrier
 5 *
 6 * @Author: Mr.Tk
 7 * @Date: 2019-03-10 15:39
 8 */
 9 public class CyclicBarrierDemo {
10 
11     //创建一个线程池
12     private static final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(5,10,60, TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>());
13 
14     //当拦截线程数量到达5，优先执行barrierAction参数线程方法，其次执行被拦截线程
15     private static final CyclicBarrier cyclicBarrier = new CyclicBarrier(5,new Runnable(){
16         @Override
17         public void run() {
18             System.out.println("裁判：预备... 跑！啾啪...枪声响起");
19         }
20     });
21 
22     private static class TestThread extends Thread {
23 
24         private String  name ;
25 
26         public TestThread(String name){
27             this.name=name;
28         }
29 
30         @Override
31         public void run() {
32             System.out.println("各就各位："+name);
33             try {
34                 Thread.sleep(1000);
35                 //拦截线程
36                 cyclicBarrier.await();
37                 Thread.sleep(1000);
38                 System.out.println("到达终点："+name);
39             } catch (InterruptedException e) {
40                 e.printStackTrace();
41             } catch (BrokenBarrierException e) {
42                 e.printStackTrace();
43             }
44         }
45     }
46 
47     public static void main(String[] args) {
48         String[] str = {"运动员1","运动员2","运动员3","运动员4","运动员5"};
49         for (int i=0 ; i<5 ;i++){
50             //调用线程
51             threadPool.execute(new TestThread(str[i]));
52 
53         }
54         try {
55             Thread.sleep(3000);//str.length*1000
56             System.out.println("裁判宣布成绩...hello world............");
57         } catch (InterruptedException e) {
58             e.printStackTrace();
59         }
60 
61 
62     }
63 }

 

"小伙伴"code非线程池版

import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;

/**
* 非线程池操作
*
* @Author: Mr.Tk
* @Date: 2019-03-10 17:19
*/
public class CyclicBarrierDemo2 implements Runnable{

    //封装私有并发工具
    private CyclicBarrier cyclicBarrier;

    public CyclicBarrierDemo2(CyclicBarrier cyclicBarrier) {
        this.cyclicBarrier = cyclicBarrier;
    }

    @Override
    public void run() {
        try {
            System.out.println("ready");
            Thread.sleep(1000);
            cyclicBarrier.await();
            System.out.println("结束");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (BrokenBarrierException e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        //线程就绪5个后执行，线程方法run
        CyclicBarrierDemo2 cyclicBarrier = new CyclicBarrierDemo2(new CyclicBarrier(5, new Runnable() {
            @Override
            public void run() {
                System.out.println("裁判：预备... 跑！啾啪...枪声响起");
            }
        }));
        Thread thread1 = new Thread(cyclicBarrier);
        Thread thread2 = new Thread(cyclicBarrier);
        Thread thread3 = new Thread(cyclicBarrier);
        Thread thread4 = new Thread(cyclicBarrier);
        Thread thread5 = new Thread(cyclicBarrier);
        thread1.start();
        thread2.start();
        thread3.start();
        thread4.start();
        thread5.start();

        //子线程执行完毕让每个子线程加入到主线程中。最后主线继续运行下去
        thread1.join();
        thread5.join();
        thread2.join();
        thread3.join();
        thread4.join();


        System.out.println("主线程执行完毕死亡，hello world ...");

    }

}

 

总结：从上代码看出，线程池的sleep方法暂时解决的主线程在自线程执行结束后，可以继续运行。但是就数据量大的处理来看，我门无法管理睡眠时间的长短。不能确定准确的睡眠时间所以存在风险。

　　其二，如果我门不使用线程池来管理线程，面临的是内存性能的消耗，对于系统来说也是一笔开销。Condition这个锁对象是“小伙伴”提供给我的思路，暂时没有深入问题暂时遗留，有了解的可以私信讨论。

　　个人设想：使用线程池管理线程时，将主线程挂起，将每个执行完毕的子线程（getCurrentThread()当前线程）执行等待方法（wait()）,之后使用notifyAll()方法再将线程唤醒到主线程继续执行，这里又涉及到主线程等待后，子线程也全班等待后，谁将唤醒主线程的问题？如果让最后一个挂起线程唤醒主线程，具体怎么确定哪一个线程是执行到最后的？问题遗留。。。ing

注：不过就目前来看，该面试题是解决了... 之后会继续深入了解