描述

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

CountDownLatch

以下代碼為例：主線程阻塞直到CountDownLatch計數(shù)為0。語義是一個線程等待一組線程執(zhí)行完畢后再繼續(xù)執(zhí)行。參考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例子。執(zhí)行流程：CyclicBarrier以await()將線程分為兩部分。第一部分先執(zhí)行，線程阻塞在await()直到CyclicBarrier計數(shù)減為0。此時如果CyclicBarrier有指定Runnable，則選取一個線程執(zhí)行完Runnable。再執(zhí)行第二部分。并且CyclicBarrier是可復用的。

語義是一個屏障，每個線程都阻塞到這個屏障上。屏障消除所有線程繼續(xù)執(zhí)行。

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時觸發(fā)CyclicBarrier中指定的Runnable，該Runnable執(zhí)行完畢。
                    * 再繼續(xù)執(zhí)行完原來的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

語義：對于一個共享資源只能有固定數(shù)量線程訪問?？梢杂糜诔鼗夹g(shù)。參考semaphore。

可以看到下面的輸出內(nèi)容，每隔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。語義：兩個線程通過共享容器交換數(shù)據(jù)。

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/