java并发编程- 5 - 线程池下join()的替代方案：CountDownLatch、CyclicBarrier

假设有一个场景：生产汽车分为了三步：制造车身、制造轮子、组装车身和轮子。单线程下我们的代码：

@RunWith(SpringRunner.class)
@SpringBootTest
@Slf4j
public class CarDemo {

     @Test
    public void madeCarTest1(){
        StopWatch stopWatch=new StopWatch();
        stopWatch.start();
        int n=5;
        for(int i =0;i<n;i++){
            //单线程
            log.info("开始制造");
            String body=madeBody();
            log.info(body);
            String wheels=madeWheels();
            log.info(wheels);
            String car=madeCar(body,wheels);
            log.info(car);
        }
        stopWatch.stop();
        log.info("制造"+n+"台汽车耗时"+stopWatch.getTotalTimeSeconds()+"秒");
    }



    private String madeBody(){
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "车身";
    }

    private String madeWheels(){
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "轮子";
    }

    private String madeCar(String body,String wheels){
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return body+wheels;
    }
}

image.png

单线程下，车身、轮子和组装都是穿行的，耗时15秒。
咱们改成多线程：

private String body;
    private String wheels;

    @Test
    public void madeCarTest2() throws InterruptedException {
        //多线程
        StopWatch stopWatch=new StopWatch();
        stopWatch.start();
        int n=5;
        for(int i =0;i<n;i++){


            Thread t1=new Thread(()->{
                body=madeBody();
                log.info(body);
            });
            Thread t2=new Thread(()->{
                wheels=madeWheels();
                log.info(wheels);
            });
            t1.start();
            t2.start();
            t1.join();
            t2.join();

            String car=madeCar(body,wheels);
            log.info(car);
        }
        stopWatch.stop();
        log.info("制造"+n+"台汽车耗时"+stopWatch.getTotalTimeSeconds()+"秒");


    }

    private String madeBody(){
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "车身";
    }

    private String madeWheels(){
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "轮子";
    }

    private String madeCar(String body,String wheels){
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return body+wheels;
    }

image.png

改成多线程以后，制造车身和制造轮子都可以并行进行了，耗时10s 提升了30%。
但是这里有个小缺陷，我们的线程都是new的，重复的创建线程太耗资源了，我们应该用线程池对线程重复利用。但是一旦改成线程池，线程就不会真正结束，所以join()方法就失效了。java为我们提供了解决方法：CountDownLatch。它的原理也很简单，就是有一个计数器，countDown()方法计数，await可以让线程等待，直到计数器次数达到目标值：

 Executor executor= Executors.newFixedThreadPool(2);//创建固定线程数为2的线程池


    @Test
    public void madeCarTest3() throws InterruptedException {
        //多线程
        StopWatch stopWatch=new StopWatch();
        stopWatch.start();
        int n=5;
        for(int i =0;i<n;i++){
            CountDownLatch countDownLatch=new CountDownLatch(2);//创建一个大小为2的计数器
            executor.execute(()->{
             
                    body=madeBody();
                    log.info(body);
                    countDownLatch.countDown();
             
            });
            executor.execute(()->{
              
                    wheels=madeWheels();
                    log.info(wheels);
                    countDownLatch.countDown();
               
            });
            countDownLatch.await();
            String car=madeCar(body,wheels);
            log.info(car);
        }
        stopWatch.stop();
        log.info("制造"+n+"台汽车耗时"+stopWatch.getTotalTimeSeconds()+"秒");
    }


    private String madeBody(){
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "车身";
    }

    private String madeWheels(){
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "轮子";
    }

    private String madeCar(String body,String wheels){
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return body+wheels;
    }

image.png

可以看到，比原来快了0.01秒多。
这个程序已经是最优了的吗？不，还可以继续优化。因为madeCar()的时候，我们还可以继续制造下一个的轮子和车身。你使用CountDownLatch也可以完成这部分工作，但是java为我们提供更方便的CyclicBarrier，CyclicBarrier 在达到期望数值的时候，回调一个方法，并且把数值重置为初始值：

Vector<String> bodys=new Vector<String>();
    Vector<String> wheelss=new Vector<String>();
    ExecutorService executor1= Executors.newFixedThreadPool(3);//创建固定线程数为2的线程池

    CyclicBarrier barrier=new CyclicBarrier(3,()->{
        executor1.execute(()->{
            String body=bodys.remove(0);//拿第一个body
            String wheels=wheelss.remove(0);//拿第一个wheels
            String car=madeCar(body,wheels);
            log.info(car);
        });
    });//创建一个计数器，当数值达到2的时候，调一次
    @Test
    public void madeCarTest4() throws InterruptedException {


        //多线程
        StopWatch stopWatch=new StopWatch();
        stopWatch.start();
        int n=5;
        for(int i =0;i<n;i++){

            executor1.execute(()->{

                    body=madeBody();
                    bodys.add(body);
                    log.info(body);
                    try {
                        barrier.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) {
                        e.printStackTrace();
                    }

            });
            executor1.execute(()->{

                    wheels=madeWheels();
                    wheelss.add(wheels);
                    log.info(wheels);
                    try {
                        barrier.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) {
                        e.printStackTrace();
                    }

            });
            try {
                barrier.await();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
        executor1.shutdown();//关闭线程池，但是会等线程执行完毕
        boolean stat=executor1.awaitTermination(2,TimeUnit.HOURS);//挂起线程，直到线程池关闭
        if(stat){
            log.info("所有线程执行完毕");
        }else {
            log.info("超时或者被中断");
        }

        stopWatch.stop();
        log.info("制造"+n+"台汽车耗时"+stopWatch.getTotalTimeSeconds()+"秒");
    }

    private String madeBody(){
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "车身";
    }

    private String madeWheels(){
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "轮子";
    }

    private String madeCar(String body,String wheels){
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return body+wheels;
    }

image.png

我们改成了这种类似于生产-消费模式后，速度更快了。因为在组装车子的时候，下一个车子的车身和轮子已经开始制造了。

最后，上面的代码其实有个问题：CyclicBarrier 的回调方法其实不应该跟其他公用一个线程池，应该单独使用一个长度为1的线程池。大家想想为什么？
下一章 java并发编程 - 6 - 并发容器