理解Java-CountDownLatch应用

發表於 | 分類於 | | 閱讀次數:

CountDownLatch说明

CountDownLatch是一个同步辅助类,这个类能够使一个线程等待其他线程完成各自的工作后再执行。例如，应用程序的主线程希望在负责启动框架服务的线程已经启动所有的框架服务之后再执行。

CountDownLatch实现概要

CountDownLatch是通过一个计数器来实现的，计数器的初始值为线程的数量。每当一个线程完成了自己的任务后，计数器的值就会减1。当计数器值到达0时，它表示所有的线程已经完成了任务，然后在闭锁上等待的线程就可以恢复执行任务。

CountDownLatch多线程应用

列表循环遍历(耗时8697ms)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
@Test
public void optimizeListV1(){
    long start = System.currentTimeMillis();
    try {
        final List<String> lists = Arrays.asList("aa", "bb", "cc", "dd", "ee");
        for(int i=0; i<lists.size(); i++){
            if(i == 2){
                Thread.sleep(3000);
            }
            Thread.sleep(1000);
        }
        System.out.println("聚合完成");
    }catch (Exception e){

    }finally {
        MockTimeUtil.mockInvokeTime("循环列表场景模拟:", start);
    }
}

多线程聚合列表(耗时4671ms)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
@Test
public void optimizeList(){
    long start = System.currentTimeMillis();
    try {
       ExecutorService ex = Executors.newFixedThreadPool(5);
       final List<String> lists = Arrays.asList("aa", "bb", "cc", "dd", "ee");
       final CountDownLatch latch =  new CountDownLatch(lists.size());
       for(int i=0; i<lists.size(); i++){
           final int tmp = i;
           ex.submit(new Callable<Object>() {
               @Override
               public Object call() throws Exception {
                   if(tmp == 2){
                       Thread.sleep(3000);
                   }
                   Thread.sleep(1000);
                   latch.countDown();
                   return null;
               }
           });
       }
       //latch.await();
       latch.await(3500, TimeUnit.MILLISECONDS);
       System.out.println("聚合完成");
   }catch (Exception e){

   }finally {
       MockTimeUtil.mockInvokeTime("线程列表场景模拟:", start);
   }
}

CountDownLatch方法说明

CountDownLatch源码

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
public class CountDownLatch {
    /**
     * Synchronization control For CountDownLatch.
     * Uses AQS state to represent count.
     */
    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;

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

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

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

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

    public long getCount() {
        return sync.getCount();
    }

    public String toString() {
        return super.toString() + "[Count = " + sync.getCount() + "]";
    }
}

方法说明:

  • countDown()：当前线程调用此方法，则计数减一
  • await(): 调用此方法会一直阻塞当前线程，直到计时器的值为0