Semaphore

Semaphore也就是信号量，提供了资源数量的并发访问控制

image.png

  1. public class Main {
  2.     public static void main(String[] args) {
  3.         Semaphore semaphore = new Semaphore(1);
  5.         for (int i = 0; i < 5; i++) {
  6.             new MyThread("学生-" + (i + 1), semaphore).start();
  7.         }
  9.     }
  10. }
  11. class MyThread extends Thread {
  12.     private final Semaphore semaphore;
  13.     private final Random random = new Random();
  15.     public MyThread(String name, Semaphore semaphore) {
  16.         super(name);
  17.         this.semaphore = semaphore;
  18.     }
  21.     @Override
  22.     public void run() {
  24.         try {
  25.             // 获取信标：抢座
  26.             semaphore.acquire();
  27.             // 抢到之后开始写作业
  28.             System.out.println(Thread.currentThread().getName() + " - 抢到了座位，开始写作业");
  29.             Thread.sleep(random.nextInt(1000));
  30.             System.out.println(Thread.currentThread().getName() + " - 作业写完，腾出座位");
  31.         } catch (InterruptedException e) {
  32.             e.printStackTrace();
  33.         }
  35.         // 释放信标：腾出座位
  36.         semaphore.release();
  37.     }
  38. }

image.png

CountDownLatch

假设一个主线程要等待5个 Worker 线程执行完才能退出，可以使用CountDownLatch来实现：

image.png
image.png

  1. public class CountDownLatchClient {
  2.     public static void main(String[] args) throws InterruptedException {
  3.         CountDownLatch latch = new CountDownLatch(5);
  4.         for (int i = 0; i < 5; i++) {
  5.             new MyThread("线程" + (i + 1), latch).start();
  6.         }
  7.         // main线程等待
  8.         latch.await();
  9.         System.out.println("main线程执行结束");
  10.     }
  11. }
  13. class MyThread extends Thread {
  14.     private final CountDownLatch latch;
  15.     private final Random random = new Random();
  17.     public MyThread(String name, CountDownLatch latch) {
  18.         super(name);
  19.         this.latch = latch;
  20.     }
  22.     @Override
  23.     public void run() {
  24.         try {
  25.             Thread.sleep(random.nextInt(2000));
  26.         } catch (InterruptedException e) {
  27.             e.printStackTrace();
  28.         }
  29.         System.out.println(Thread.currentThread().getName() + " - 执行完毕");
  30.         // latch计数减一
  31.         latch.countDown();
  32.     }
  33. }
  35. "C:\Program Files\Java\jdk1.8.0_181\bin\java.exe" "-javaagent:E:\software\dev_tool\idea\exec\IntelliJ IDEA 2021.1.2\lib\idea_rt.jar=55649:E:\software\dev_tool\idea\exec\IntelliJ IDEA 2021.1.2\bin" -Dfile.encoding=UTF-8 -classpath "C:\Program Files\Java\jdk1.8.0_181\jre\lib\charsets.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\deploy.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\ext\access-bridge-64.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\ext\cldrdata.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\ext\dnsns.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\ext\jaccess.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\ext\jfxrt.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\ext\localedata.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\ext\nashorn.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\ext\sunec.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\ext\sunjce_provider.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\ext\sunmscapi.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\ext\sunpkcs11.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\ext\zipfs.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\javaws.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\jce.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\jfr.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\jfxswt.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\jsse.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\management-agent.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\plugin.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\resources.jar;C:\Program Files\Java\jdk1.8.0_181\jre\lib\rt.jar;D:\dev_need\workspace\concurrent-java\basic-demo\target\classes;C:\Users\wpp25\.m2\repository\junit\junit\4.12\junit-4.12.jar;C:\Users\wpp25\.m2\repository\org\hamcrest\hamcrest-core\1.3\hamcrest-core-1.3.jar" com.wpp.concurrent.util.countdownlatchdemo.CountDownLatchClient
  36. 线程3 - 执行完毕
  37. 线程2 - 执行完毕
  38. 线程1 - 执行完毕
  39. 线程4 - 执行完毕
  40. 线程5 - 执行完毕
  41. main线程执行结束

image.png

await()实现分析

await()调用的是AQS 的模板方法，这个方法在前面已经介绍过。 CountDownLatch.Sync重新实现了tryAccuqireShared方法：

  • CountDownLatch

    1. public boolean await(long timeout, TimeUnit unit) throws InterruptedException {
    2.        // AQS的模板方法
    3.       return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    4. }

  • AbstractQueuedSynchronizer

    1. public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException {
    2.       if (Thread.interrupted())
    3.           throw new InterruptedException();
    4.       return tryAcquireShared(arg) >= 0 ||
    5.           doAcquireSharedNanos(arg, nanosTimeout);
    6.   }

  • CountDownLatch

    1. protected int tryAcquireShared(int acquires) {
    2.           return (getState() == 0) ? 1 : -1;
    3.       }

    从tryAcquireShared(…)方法的实现来看，只要state != 0，调用await()方法的线程便会被放入AQS
    的阻塞队列，进入阻塞状态。

    countDown()实现分析

  • CountDownLatch

    1. public void countDown() {
    2.    // AQS的模板方法
    3.       sync.releaseShared(1);
    4.   }

  • AbstractQueuedSynchronizer

    1. public final boolean releaseShared(int arg) {
    2.       if (tryReleaseShared(arg)) {
    3.           doReleaseShared();
    4.           return true;
    5.       }
    6.       return false;
    7.   }

  • CountDownLatch

    1. protected boolean tryReleaseShared(int releases) {
    2.           // Decrement count; signal when transition to zero
    3.           for (;;) {
    4.               int c = getState();
    5.               if (c == 0)
    6.                   return false;
    7.               int nextc = c-1;
    8.               if (compareAndSetState(c, nextc))
    9.                   return nextc == 0;
    10.           }
    11.       }

    countDown()调用的AQS的模板方法releaseShared()，里面的tryReleaseShared(…)由
    CountDownLatch.Sync实现。从上面的代码可以看出，只有state=0，tryReleaseShared(…)才会返回
    true，然后执行doReleaseShared(…)，一次性唤醒队列中所有阻塞的线程。

总结：由于是基于AQS阻塞队列来实现的，所以可以让多个线程都阻塞在state=0条件上，通过
countDown()一直减state，减到0后一次性唤醒所有线程。如下图所示，假设初始总数为M，N个线程
await()，M个线程countDown()，减到0之后，N个线程被唤醒
image.png

CyclicBarrier

该类用于协调多个线程同步执行操作的场合。

使用场景：10个工程师一起来公司应聘，招聘方式分为笔试和面试。
首先，要等人到齐后，开始笔试；笔试结束之后，再一起参加面试。
把10个人看作10个线程，10个线程之间的同步过程如下图所示：
image.png
在整个过程中，有2个同步点：第1个同步点，要等所有应聘者都到达公司，再一起开始笔试；第2
个同步点，要等所有应聘者都结束笔试，之后一起进入面试环节。

  1. public class CyclicBarrierClient {
  2.     public static void main(String[] args) {
  3. //        CyclicBarrier barrier = new CyclicBarrier(5);
  4.         CyclicBarrier barrier = new CyclicBarrier(5, () -> {
  5.             System.out.println("该阶段结束");
  6.         });
  7.         for (int i = 0; i < 5; i++) {
  8.             new MyThread("线程-" + (i + 1), barrier).start();
  9.         }
  10.     }
  11. }
  12. class MyThread extends Thread {
  14.     private final CyclicBarrier barrier;
  15.     private final Random random = new Random();
  17.     public MyThread(String name, CyclicBarrier barrier) {
  18.         super(name);
  19.         this.barrier = barrier;
  20.     }
  22.     @Override
  23.     public void run() {
  24.         try {
  25.             System.out.println(Thread.currentThread().getName() + " - 向公司出发");
  26.             Thread.sleep(random.nextInt(5000));
  27.             System.out.println(Thread.currentThread().getName() + " - 已经到达公司");
  28.             // 等待其他线程该阶段结束
  29.             barrier.await();
  31.             System.out.println(Thread.currentThread().getName() + " - 开始笔试");
  32.             Thread.sleep(random.nextInt(5000));
  33.             System.out.println(Thread.currentThread().getName() + " - 笔试结束");
  34.             // 等待其他线程该阶段结束
  35.             barrier.await();
  37.             System.out.println(Thread.currentThread().getName() + " - 开始面试");
  38.             Thread.sleep(random.nextInt(5000));
  39.             System.out.println(Thread.currentThread().getName() + " - 面试结束");
  41.         } catch (InterruptedException e) {
  42.             e.printStackTrace();
  43.         } catch (BrokenBarrierException e) {
  44.             e.printStackTrace();
  45.         }
  46.     }
  47. }

Exchanger

Exchanger用于线程之间交换数据

  1. public class ExchangeClient {
  2.     private static final Random random = new Random();
  3.     public static void main(String[] args) {
  4.         // 建一个多线程共用的exchange对象
  5.         // 把exchange对象传给3个线程对象。每个线程在自己的run方法中调用exchange，把自己的数据作为参数
  6.         // 传递进去，返回值是另外一个线程调用exchange传进去的参数
  7.         Exchanger<String> exchanger = new Exchanger<>();
  9.         new Thread("线程1") {
  10.             @Override
  11.             public void run() {
  12.                 while (true) {
  13.                     try {
  14.                         // 如果没有其他线程调用exchange，线程阻塞，直到有其他线程调用exchange为止。
  15.                         String otherData = exchanger.exchange("交换数据1");
  16.                         System.out.println(Thread.currentThread().getName() + "得到<==" + otherData);
  17.                         Thread.sleep(random.nextInt(2000));
  18.                     } catch (InterruptedException e) {
  19.                         e.printStackTrace();
  20.                     }
  21.                 }
  22.             }
  23.         }.start();
  25.         new Thread("线程2") {
  26.             @Override
  27.             public void run() {
  28.                 while (true) {
  29.                     try {
  30.                         String otherData = exchanger.exchange("交换数据2");
  31.                         System.out.println(Thread.currentThread().getName() + "得到<==" + otherData);
  32.                         Thread.sleep(random.nextInt(2000));
  33.                     } catch (InterruptedException e) {
  34.                         e.printStackTrace();
  35.                     }
  36.                 }
  37.             }
  38.         }.start();
  40.         new Thread("线程3") {
  41.             @Override
  42.             public void run() {
  43.                 while (true) {
  44.                     try {
  45.                         String otherData = exchanger.exchange("交换数据3");
  46.                         System.out.println(Thread.currentThread().getName() + "得到<==" + otherData);
  47.                         Thread.sleep(random.nextInt(2000));
  48.                     } catch (InterruptedException e) {
  49.                         e.printStackTrace();
  50.                     }
  51.                 }
  52.             }
  53.         }.start();
  54.     }
  55. }

Phaser

从JDK7开始，新增了一个同步工具类Phaser，其功能比CyclicBarrier和CountDownLatch更加强大。

  1. public class FunOfCountDownLatch {
  2.     public static void main(String[] args) {
  3.         Phaser phaser = new Phaser(5);
  5.         for (int i = 0; i < 5; i++) {
  6.             new Thread("线程-" + (i + 1)) {
  7.                 private final Random random = new Random();
  8.                 @Override
  9.                 public void run() {
  10.                     System.out.println(getName() + " - 开始运行");
  11.                     try {
  12.                         Thread.sleep(random.nextInt(1000));
  13.                     } catch (InterruptedException e) {
  14.                         e.printStackTrace();
  15.                     }
  16.                     System.out.println(getName() + " - 运行结束");
  17.                     phaser.arrive();
  18.                 }
  19.             }.start();
  20.         }
  21.         System.out.println("线程启动完毕");
  22.         System.out.println(phaser.getPhase());
  23. //        phaser.awaitAdvance(phaser.getPhase());
  24.         phaser.awaitAdvance(0);
  25.         System.out.println("线程运行结束");
  26.     }
  27. }
  1. public class FunOfCyclicBarrier {
  2.     public static void main(String[] args) {
  3.         Phaser phaser = new Phaser(5);
  5.         for (int i = 0; i < 5; i++) {
  6.             new MyThread("线程-" + (i + 1), phaser).start();
  7.         }
  9.         phaser.awaitAdvance(0);
  10.         System.out.println("运行结束");
  11.     }
  12. }
  14. class MyThread extends Thread {
  16.     private final Phaser phaser;
  17.     private final Random random = new Random();
  19.     public MyThread(String name, Phaser phaser) {
  20.         super(name);
  21.         this.phaser = phaser;
  22.     }
  24.     @Override
  25.     public void run() {
  26.         System.out.println(getName() + " - 开始向公司出发");
  27.         slowly();
  28.         System.out.println(getName() + " - 已经到达公司");
  29.         // 到达同步点，等待其他线程
  30.         phaser.arriveAndAwaitAdvance();
  32.         System.out.println(getName() + " - 开始笔试");
  33.         slowly();
  34.         System.out.println(getName() + " - 笔试结束");
  35.         // 到达同步点，等待其他线程
  36.         phaser.arriveAndAwaitAdvance();
  38.         System.out.println(getName() + " - 开始面试");
  39.         slowly();
  40.         System.out.println(getName() + " - 面试结束");
  41.     }
  43.     private void slowly() {
  44.         try {
  45.             Thread.sleep(random.nextInt(1000));
  46.         } catch (InterruptedException e) {
  47.             e.printStackTrace();
  48.         }
  49.     }
  50. }