并发工具类CountDownLatch源码分析

傷城~ 2022-05-24 02:55 278阅读 0赞

> 同步工具类可以使任何一种对象，只要该对象可以根据自身的状态来协调控制线程的控制流。阻塞队列可以作为同步工具类，其他类型的同步工具类还包括：信号量（Semaphore）、栅栏（Barrier）、闭锁（Latch）以及交换器（Exchanger）。 –《Java并发编程实战》

[Semaphore、CyclicBarrier 、CountDownLatch][Semaphore_CyclicBarrier _CountDownLatch]

## 简介 ##

> **CountDownLatch 允许一个或多个线程等待其他线程先完成工作。**

CountDownLatch是通过“共享锁”实现的。在创建CountDownLatch中时，会传递一个int类型参数count，该参数是“锁计数器”的初始状态，表示该“共享锁”最多能被count个线程同时获取。当某线程调用该CountDownLatch对象的await()方法时（当前线程阻塞），该线程会等待“共享锁”可用时，才能获取“共享锁”进而继续运行。而“共享锁”可用的条件，就是“锁计数器”的值为0。而“锁计数器”的初始值为count，每当一个线程调用该CountDownLatch对象的countDown()方法时，才将“锁计数器”-1；**通过这种方式，必须有count个线程调用**countDown()**之后，“锁计数器”才为0，而前面提到的等待线程才能继续运行。**

## 说明 ##

// downLatchz 最多可以阻塞8个线程
    int count=8;
    CountDownLatch downLatch=new CountDownLatch(count);
    
    // 在某个线程里调用
    downLatch.await();
    // 则阻塞当前线程，downLatch.await();的业务先不执行
    
    // 可以在任意一个线程中调用downLatch.countDown();，即count-1操作
    downLatch.countDown();
    // 当count减为0时，前面所有被downLatch.await();阻塞的线程都可以执行后面的业务操作

## 应用场景 ##

### 场景一 ###

> **有一个任务想要往下执行，但必须要等到其他的任务执行完毕后才可以继续往下执行。假如我们这个想要继续往下执行的任务调用一个CountDownLatch对象的await()方法，其他的任务执行完自己的任务后调用同一个CountDownLatch对象上的countDown()方法，这个调用await()方法的任务将一直阻塞等待，直到这个CountDownLatch对象的计数值减到0为止。**

#### DEMO代码 ####

public class MainTest { 
        public enum EnumTest { 
            发动机, 底盘, 车身, 其他;// 线程名称
        }
    
        public static void main(String[] args) { 
        	// 计数器为3
            final CountDownLatch countDownLatch = new CountDownLatch(3);
            new Thread(new Runnable() { 
                @Override
                public void run() { 
                    System.out.println("主线程:正在等待车辆质检");
                    try { 
                        // 阻塞当前线程
                        countDownLatch.await();
                        System.out.println("发动机、底盘、车身、其他,已配置好, 车辆出厂啦");
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }).start();
            for (EnumTest threadName : EnumTest.values()) { 
                final String tN = String.valueOf(threadName);
                new Thread(new Runnable() { 
                    @Override
                    public void run() { 
                        System.out.println(tN + " 正在配置");
                        try { 
                            Thread.sleep(100);
                        } catch (Exception e) { 
                            e.printStackTrace();
                        }
                        System.out.println(tN + " 安装完成");
                        // 减一操作
                        countDownLatch.countDown();
                    }
                }).start();
            }
    
        }
    }

## 模拟高并发 ##

package com.sto.tmsapp.util;
    
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.atomic.AtomicInteger;
    
    public class CountDownLatchTest { 
        public static void concurrenceTest() { 
            /** * 模拟高并发情况代码 */
            final AtomicInteger atomicInteger = new AtomicInteger(0);
            // 相当于计数器，当所有都准备好了，再一起执行，模仿多并发，保证并发量
            final CountDownLatch latch = new CountDownLatch(1000);
            // 保证所有线程执行完了再打印atomicInteger的值
            final CountDownLatch countDownLatch = new CountDownLatch(1000);
            ExecutorService executorService = Executors.newFixedThreadPool(10);
            try { 
                for (int i = 0; i < 1000; i++) { 
                    executorService.submit(new Runnable() { 
                        @Override
                        public void run() { 
                            try { 
                                //一直阻塞当前线程，直到计时器的值为0,保证同时并发
                                latch.await();
                            } catch (InterruptedException e) { 
                                e.printStackTrace();
                            }
                            // 高并发代码块
                            //每个线程增加1000次，每次加1
                            for (int j = 0; j < 1000; j++) { 
                                atomicInteger.incrementAndGet();
                            }
                            countDownLatch.countDown();
                        }
                    });
                    latch.countDown();
                }
    
                // 保证所有线程执行完
                countDownLatch.await();
                System.out.println(atomicInteger);
                executorService.shutdown();
            } catch (Exception e) { 
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args) throws InterruptedException { 
            concurrenceTest();
        }
    }

## 源码分析 ##

### 类图 ###

![这里写图片描述][70]

### 构造函数 ###

> 构造函数需要传入一个大于的零的数

public CountDownLatch(int count) { 
            if (count < 0) throw new IllegalArgumentException("count < 0");
            this.sync = new Sync(count);
        }

### 内部类Sync ###

在CountDownLatch内部，有一个Sync的同步器，它继承自java.util.concurrent包中各种同步工具共用的AbstractQueuedSynchronizer，其实现如下：

private static final class Sync extends AbstractQueuedSynchronizer { 
          private static final long serialVersionUID = 4982264981922014374L;
    
          Sync(int count) { 
              setState(count);
          }
    
          int getCount() { 
              return getState();
          }
    
          protected int tryAcquireShared(int acquires) { 
              return (getState() == 0) ? 1 : -1;
          }
    
          protected boolean tryReleaseShared(int releases) { 
              // Decrement count; signal when transition to zero
              for (;;) { 
                  int c = getState();
                  if (c == 0)
                      return false;
                  int nextc = c-1;
                  if (compareAndSetState(c, nextc))
                      return nextc == 0;
              }
          }
      }

### 成员变量 ###

private final Sync sync;//继承自AQS

### await()方法 ###

> await()方法的核心作用是，让当前线程阻塞，直到latch的count值更改为0，或者当前线程被interrupted。

public void await() throws InterruptedException { 
         sync.acquireSharedInterruptibly(1);
     }
    
     public boolean await(long timeout, TimeUnit unit)
         throws InterruptedException { 
         return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
     }

### countDown()方法 ###

> countDown方法则调用AQS的releaseShared方法，释放共享锁，也就是每次将state状态每次减一，直到减到0，则唤醒队列中的所有节点（线程）。

public void countDown() { 
         sync.releaseShared(1);
     }
    
     /** * @return the current count */
     public long getCount() { 
         return sync.getCount();
     }
    
     /** * @return a string identifying this latch, as well as its state */
     public String toString() { 
         return super.toString() + "[Count = " + sync.getCount() + "]";
     }

[Semaphore_CyclicBarrier _CountDownLatch]: https://blog.csdn.net/jiangqiongfeng/article/details/79456588
[70]: /images/20220524/3da9568036ca406a8bd1a7f0f95d6753.png