描述

本文描述了JUC中CountDownLatch、CyclicBarrier、Semaphore、Exchanger工具类使用方式。

CountDownLatch

以下代码为例：主线程阻塞直到CountDownLatch计数为0。语义是一个线程等待一组线程执行完毕后再继续执行。参考countdownlatch例子。

public static void main(String[] args) {

      int count = 3;
      CountDownLatch latch = new CountDownLatch(count);

      for (int i = 0; i < count; ++i) {
          final int a = i;
          new Thread(() -> {
              try {
                  Thread.sleep(1000 * a);
              } catch (InterruptedException e) {
                  e.printStackTrace();
              }
              System.out.println(String.format("%s:%s:action", Thread.currentThread().getName(), a));
              latch.countDown();
          }).start();
      }

      try {
          latch.await();
      } catch (InterruptedException e) {
          e.printStackTrace();
      }

      System.out.println(String.format("%s:action finished", Thread.currentThread().getName()));
  }

输出

Thread-0:0:action
Thread-1:1:action
Thread-2:2:action
main:action finished

CyclicBarrier

参考cyclicbarrier例子。执行流程：CyclicBarrier以await()将线程分为两部分。第一部分先执行，线程阻塞在await()直到CyclicBarrier计数减为0。此时如果CyclicBarrier有指定Runnable，则选取一个线程执行完Runnable。再执行第二部分。并且CyclicBarrier是可复用的。

语义是一个屏障，每个线程都阻塞到这个屏障上。屏障消除所有线程继续执行。

public static void main(String[] args) {

      int count = 3;
      CyclicBarrier barrier = new CyclicBarrier(count,()->{
          System.out.println(String.format("%s:complete finished", Thread.currentThread().getName()));
      });

      for (int i = 0; i < count; i++) {

          final int a = i;
          new Thread(() -> {

              try {
                  Thread.sleep(1000 * (a+1));
              } catch (InterruptedException e) {
                  e.printStackTrace();
              }

              System.out.println(String.format("%s:%s:wait", Thread.currentThread().getName(), barrier.getParties() - barrier.getNumberWaiting() - 1));

              try {

                  /**
                    * 线程阻塞到此，直到CyclicBarrier中count减少到0；
                    * 减少到0时触发CyclicBarrier中指定的Runnable，该Runnable执行完毕。
                    * 再继续执行完原来的Runnable
                    */
                  barrier.await();
              } catch (InterruptedException | BrokenBarrierException e) {
                  e.printStackTrace();
              }

              System.out.println(String.format("%s:%s:action", Thread.currentThread().getName(), barrier.getParties()));

          }, "Thread1-" + a).start();
      }


      // CyclicBarrier可以重复使用
      for (int i = 0; i < count; i++) {
          final int a = i;
          new Thread(() -> {

              try {
                  Thread.sleep(1000 * (a + 4));
              } catch (InterruptedException e) {
                  e.printStackTrace();
              }

              System.out.println(String.format("%s:%s:wait", Thread.currentThread().getName(), barrier.getParties() - barrier.getNumberWaiting() - 1));

              try {
                  barrier.await();
              } catch (InterruptedException | BrokenBarrierException e) {
                  e.printStackTrace();
              }

              System.out.println(String.format("%s:%s:action", Thread.currentThread().getName(), barrier.getParties()));

          }, "Thread2-" + a).start();
      }
  }

输出

Thread1-0:2:wait
Thread1-1:1:wait
Thread1-2:0:wait
Thread1-2:complete finished
Thread1-1:3:action
Thread1-0:3:action
Thread1-2:3:action

Thread2-0:2:wait
Thread2-1:1:wait
Thread2-2:0:wait
Thread2-2:complete finished
Thread2-2:3:action
Thread2-0:3:action
Thread2-1:3:action

Semaphore

语义：对于一个共享资源只能有固定数量线程访问。可以用于池化技术。参考semaphore。

可以看到下面的输出内容，每隔3秒就输出3条记录。

public static void main(String[] args) {
    // true表示公平锁
    Semaphore semaphore = new Semaphore(3, true);

    for (int i = 0; i < 17; i++) {
        new Thread(() -> {
            try {
                semaphore.acquire();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            System.out.println(String.format("%s:action", Thread.currentThread().getName()));

            try {
                Thread.sleep(3 * 1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            semaphore.release();
        }).start();
    }
}

输出

Thread-1:action
Thread-0:action
Thread-2:action
Thread-4:action
Thread-5:action
Thread-3:action
Thread-6:action
Thread-8:action
Thread-7:action
...

Exchanger

参考exchanger。语义：两个线程通过共享容器交换数据。

public static void main(String[] args) {

    Exchanger<String> exchanger = new Exchanger<>();

    Thread product = new Thread(() -> {

        String s = "lucy";

        try {
            System.out.println("product:" + s);

            String exchange = exchanger.exchange(s);
            System.out.println("product:" + exchange);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });

    Thread consumer = new Thread(() -> {
        String s = "lilly";

        try {
            System.out.println("consumer:" + s);

            Thread.sleep(2000);

            String exchange = exchanger.exchange(s);
            System.out.println("consumer:" + exchange);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });

    product.start();
    consumer.start();
}

输出

product:lucy
consumer:lilly

product:lilly
consumer:lucy

引用

https://java2blog.com/core-java-tutorial-for-beginners-experienced/