CountDownLatch的使用与源码分析、手写实现

古城微笑少年丶 2022-12-10 10:07 171阅读 0赞

# CountDownLatch的使用与源码分析 #

CountDownLatch俗称闭锁，它可以允许一个或多个线程等待其他线程完成指定操作后再运行。

CountDownLatch的构造函数接收一个int类型的参数作为计数器，如果你想等待N个点完成，这里就传入N。

当我们调用CountDownLatch的countDown方法时，N就会减1，CountDownLatch的await方法会阻塞当前线程，直到N变成零。

由于countDown方法可以用在任何地方，所以这里说的N个点，可以是N个线程，也可以是1个线程里的N个执行步骤（一个线程可以countDown多次）。用在多个线程时，只需要把这个CountDownLatch的引用传递到线程里即可。

## CountDownLatch中的方法 ##

<table> 
 <thead> 
  <tr> 
   <th>方法名</th> 
   <th>说明</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>CountDownLatch(int count)</td> 
   <td>构造一个以给定计数count的CountDownLatch</td> 
  </tr> 
  <tr> 
   <td>void await()</td> 
   <td>当前线程一直等到闭锁计数到零，除非线程被interrupt</td> 
  </tr> 
  <tr> 
   <td>boolean await(long timeout, TimeUnit unit)</td> 
   <td>当前线程等待一直等到闭锁计数到零，除非线程被interrupt或超时</td> 
  </tr> 
  <tr> 
   <td>void countDown()</td> 
   <td>减少闭锁的计数，如果计数达到零，唤醒所有阻塞等待的线程</td> 
  </tr> 
  <tr> 
   <td>long getCount()</td> 
   <td>返回当前计数</td> 
  </tr> 
 </tbody> 
</table>

## CountDownLatch的使用 ##

假如有这样一个需求：我们需要解析一个Excel里多个sheet的数据，此时可以考虑使用多线程，每个线程解析一个sheet里的数据，等到所有的sheet都解析完之后，程序需要提示解析完成。

package com.morris.concurrent.tool.countdownlatch.api;
    
    import lombok.extern.slf4j.Slf4j;
    
    import java.util.Random;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.TimeUnit;
    
    /** * 演示CountDownLatch闭锁的使用 */
    @Slf4j
    public class CountDownLatchDemo { 
    
        private static CountDownLatch countDownLatch = new CountDownLatch(3);
    
        public static void main(String[] args) throws InterruptedException { 
    
            new Thread(()->parse("sheet1"), "t1").start();
            new Thread(()->parse("sheet2"), "t2").start();
            new Thread(()->parse("sheet3"), "t3").start();
    
            countDownLatch.await();
            log.info("parse commplete");
        }
    
        private static void parse(String sheet) { 
            log.info("{} parse {} begin...", Thread.currentThread().getName(), sheet);
            try { 
                TimeUnit.SECONDS.sleep(new Random(System.currentTimeMillis()).nextInt(30)); // 随机休眠，模拟解析sheet耗时
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            log.info("{} parse {} end.", Thread.currentThread().getName(), sheet);
            countDownLatch.countDown();
            log.info("还有{}个sheet未解析完", countDownLatch.getCount());
        }
    }

运行结果如下：

2020-09-24 13:58:55,551  INFO [t2] (CountDownLatchDemo.java:28) - t2 parse sheet2 begin...
    2020-09-24 13:58:55,565  INFO [t1] (CountDownLatchDemo.java:28) - t1 parse sheet1 begin...
    2020-09-24 13:58:55,597  INFO [t3] (CountDownLatchDemo.java:28) - t3 parse sheet3 begin...
    2020-09-24 13:59:01,565  INFO [t1] (CountDownLatchDemo.java:34) - t1 parse sheet1 end.
    2020-09-24 13:59:01,566  INFO [t1] (CountDownLatchDemo.java:36) - 还有2个sheet未解析完
    2020-09-24 13:59:02,599  INFO [t3] (CountDownLatchDemo.java:34) - t3 parse sheet3 end.
    2020-09-24 13:59:02,599  INFO [t3] (CountDownLatchDemo.java:36) - 还有1个sheet未解析完
    2020-09-24 13:59:24,556  INFO [t2] (CountDownLatchDemo.java:34) - t2 parse sheet2 end.
    2020-09-24 13:59:24,556  INFO [t2] (CountDownLatchDemo.java:36) - 还有0个sheet未解析完
    2020-09-24 13:59:24,556  INFO [main] (CountDownLatchDemo.java:24) - parse commplete

当然也可以使用join()实现：

package com.morris.concurrent.tool.countdownlatch.api;
    
    import lombok.extern.slf4j.Slf4j;
    
    import java.util.Random;
    import java.util.concurrent.TimeUnit;
    
    /** * 使用join完成对excel的解析，与countDownLatch对比 */
    @Slf4j
    public class JoinDemo { 
    
        public static void main(String[] args) throws InterruptedException { 
    
            Thread t1 = new Thread(() -> parse("sheet1"), "t1");
            t1.start();
            Thread t2 = new Thread(() -> parse("sheet2"), "t2");
            t2.start();
            Thread t3 = new Thread(() -> parse("sheet3"), "t3");
            t3.start();
    
            t1.join();
            t2.join();
            t3.join();
            log.info("parse commplete");
        }
    
        public static void parse(String sheet) { 
            log.info("{} parse {} begin...", Thread.currentThread().getName(), sheet);
            try { 
                TimeUnit.SECONDS.sleep(new Random(System.currentTimeMillis()).nextInt(30)); // 随机休眠，模拟解析sheet耗时
            } catch (InterruptedException e) { 
                e.printStackTrace();
            }
            log.info("{} parse {} end.", Thread.currentThread().getName(), sheet);
        }
    }

运行结果如下：

2020-09-24 14:04:57,732  INFO [t2] (JoinDemo.java:30) - t2 parse sheet2 begin...
    2020-09-24 14:04:57,755  INFO [t3] (JoinDemo.java:30) - t3 parse sheet3 begin...
    2020-09-24 14:04:57,736  INFO [t1] (JoinDemo.java:30) - t1 parse sheet1 begin...
    2020-09-24 14:05:07,757  INFO [t3] (JoinDemo.java:36) - t3 parse sheet3 end.
    2020-09-24 14:05:07,758  INFO [t1] (JoinDemo.java:36) - t1 parse sheet1 end.
    2020-09-24 14:05:07,757  INFO [t2] (JoinDemo.java:36) - t2 parse sheet2 end.
    2020-09-24 14:05:07,759  INFO [main] (JoinDemo.java:26) - parse commplete

CountDownLatch与join的对比：

*  CountDownLatch比join更灵活，join必须拿到thread对象才能使用，而平常都是使用线程池，线程池中的线程不对外暴露，无法使用join。
 *  join必须等待线程运行结束后才会返回，而CountDownLatch.await()可以在线程运行一半后就返回。

## CountDownLatch的源码分析 ##

CountDownLatch的底层基于AQS实现。

### 数据结构 ###

java.util.concurrent.CountDownLatch.Sync

private static final class Sync extends AbstractQueuedSynchronizer { 
    
        Sync(int count) { 
            setState(count); // 构造时初始化count的大小
        }
    
        int getCount() { 
            return getState();
        }
    
        // await()调用此方法，不为0就会进入同步队列中等待，为0就会直接返回，往下执行
        protected int tryAcquireShared(int acquires) { 
            return (getState() == 0) ? 1 : -1;
        }
    
        // countDown()调用此方法，每调用一次state就会-1，当state=0时，会去唤醒同步队列中等待的线程
        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;
            }
        }
    }

### countDown() ###

java.util.concurrent.CountDownLatch\#countDown

public void countDown() { 
        sync.releaseShared(1);
    }

java.util.concurrent.locks.AbstractQueuedSynchronizer\#releaseShared

public final boolean releaseShared(int arg) { 
        if (tryReleaseShared(arg)) {  // state-1
            doReleaseShared(); // 唤醒同步队列中等待的线程
            return true;
        }
        return false;
    }

### await() ###

java.util.concurrent.CountDownLatch\#await()

public void await() throws InterruptedException { 
        sync.acquireSharedInterruptibly(1);
    }

java.util.concurrent.locks.AbstractQueuedSynchronizer\#acquireSharedInterruptibly

public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException { 
        if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0) // 判断state是否为0，是就会直接返回
            doAcquireSharedInterruptibly(arg); // 进入同步队列中等待，park
    }

## 自定义CountDownLatch ##

package com.morris.concurrent.tool.countdownlatch.my;
    
    import java.util.concurrent.TimeUnit;
    
    /** * 使用wait-notify实现CountDownLatch */
    public class WaitNotifyCountDownLatch { 
    
        private volatile int count;
    
        public WaitNotifyCountDownLatch(int count) { 
            this.count = count;
        }
    
        public synchronized void countDown() { 
            if (0 == --count) { 
                this.notifyAll();
            }
        }
    
        public synchronized void await() throws InterruptedException { 
            while (count > 0) { 
                this.wait();
            }
        }
        
        public synchronized void await(long timeout, TimeUnit unit) throws InterruptedException { 
            while (count > 0) { 
                this.wait(unit.toMillis(timeout));
            }
        }
    
        public int getCount() { 
            return count;
        }
    }