Java并发--CountDownLatch CyclicBarrier ReentrantLock

CountDownLatch

CountDownLatch是一个同步工具类，它允许一个或多个线程一直等待，直到其他线程的操作执行完后再执行。CountDownLatch使用一个数字count初始化，使用countDown方法使count递减，当count大于0时await方法将一直阻塞，当countDown为0时await方法将立即返回。CountDownLatch有两种典型用法，一是阻塞主线程直到所有子线程执行到某步。二是阻塞子线程直到某条件达成，下面分别是例子。

public class CountDownLatchTest {
    public static void main(String[] args) throws Exception {
        final CountDownLatch latch = new CountDownLatch(5);
        for (int k = 0; k < 5; k++) {
            final int n = k;
            new Thread(new Runnable() {
                private int id = n;
                @Override
                public void run() {
                    try {
                        System.out.println("thread " + id + " begin.");
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {
                    }
                    System.out.println("thread " + id + " run over.");
                    latch.countDown();
                }
            }).start();
        }
        latch.await();
        System.out.println("main end");
    }
}

---

public class CountDownLatchTest {
    public static void main(String[] args) throws Exception {
        final CountDownLatch latch = new CountDownLatch(1);
        for (int k = 0; k < 5; k++) {
            final int n = k;
            new Thread(new Runnable() {
                private int id = n;
                @Override
                public void run() {
                    try {
                        System.out.println("thread " + id + " begin.");
                        latch.await();
                    } catch (InterruptedException e) {
                    }
                    System.out.println("thread " + id + " run over.");
                    latch.countDown();
                }
            }).start();
        }
        latch.countDown();
        System.out.println("main end");
    }
}

---

CyclicBarrier

CyclicBarrier在初始化时规定一个数目，然后计算调用了CyclicBarrier.await()进入等待的线程数。当线程数达到了这个数目时，所有进入等待状态的线程被唤醒并继续。

CyclicBarrier初始时还可带一个Runnable的参数， 此Runnable任务在CyclicBarrier的数目达到后，所有其它线程被唤醒前被执行。通过reset函数可重置该锁。

public class CyclicBarrierTest {

    public static void main(String[] args) throws BrokenBarrierException, InterruptedException {
        CyclicBarrier barrier = new CyclicBarrier(5, new Runnable() {
            @Override
            public void run() {
                //当所有线程到达barrier时执行
                System.out.println("Barrier action");
            }
        });

        for (int k = 0; k < 4; k++) {
            final int n = k;
            new Thread(new Runnable() {
                private int id = n;
                @Override
                public void run() {
                    try {
                        System.out.println("thread " + id + " begin.");
                        TimeUnit.SECONDS.sleep(1);
                        //线程在这里等待，直到所有线程都到达barrier
                        barrier.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } catch (BrokenBarrierException e) {
                        e.printStackTrace();
                    }
                    System.out.println("thread " + id + " run over.");
                }
            }).start();
        }

        System.out.println("main end");
    }
}

---

ReentrantLock

ReentrantLock是Java concurrent包中的一个互斥锁，它可以用来替代synchronized关键字，使用更加灵活。该锁同一时间只能被一个线程拥有，即执行了lock()但还未执行unlock()的线程。

/*
 * ReentrantLock用来对一段代码上锁，可以代替synchronized关键字。
 */
class SomeClassWithLock {

    private long count1 = 0;
    private long count2 = 0;
    
    //A ReentrantLock is owned by the thread last successfully locking, but not yet unlocking it.
    private ReentrantLock lock = new ReentrantLock();

    public void test() {

        //++并非原子操作,此处未上锁
        count1++;

        lock.lock();
        try {
            //此处线程安全
            count2++;
        } finally {
            //将unlock放在finally块里面
            lock.unlock();
        }
    }

    public long get1() {
        return count1;
    }
    public long get2() {
        return count2;
    }
}

public class ReentrantLockTest {

    public static void main(String[] args) {

        final int COUNT = 10;
        final CountDownLatch startSignal = new CountDownLatch(1);
        final CountDownLatch doneSignal = new CountDownLatch(COUNT);

        final SomeClassWithLock someClass = new SomeClassWithLock();

        for (int i = 0; i < COUNT; ++i) {
            final int index = i;
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        startSignal.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    for (int j = 0; j < 1000; ++j) {
                        someClass.test();
                    }
                    System.out.println("running thread " + index);
                    doneSignal.countDown();
                }
            }).start();
        }

        startSignal.countDown();
        try {
            doneSignal.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("count1:" + someClass.get1());
        System.out.println("count2:" + someClass.get2());
    }

}

---

执行结果：

running thread 4
running thread 7
running thread 8
running thread 2
running thread 6
running thread 0
running thread 5
running thread 9
running thread 1
running thread 3
count1:9988
count2:10000

end