Java多线程编程(5)--线程间通信(上)

一.等待与通知

  某些情况下，程序要执行的操作需要满足一定的条件（下文统一将其称之为保护条件）才能执行。在单线程编程中，我们可以使用轮询的方式来实现，即频繁地判断是否满足保护条件，若不满足则继续判断，若满足则开始执行。但在多线程编程中，这种方式无疑是非常低效的。如果一个线程持续进行无意义的判断而不释放CPU，这就会造成资源的浪费；而如果定时去判断，不满足保护条件就释放CPU，又会造成频繁的上下文切换。总之，不推荐在多线程编程中使用轮询的方式。
  等待与通知是这样一种机制：当保护条件不满足时，可以将当前线程暂停；而当保护条件成立时，再将这个线程唤醒。一个线程因其保护条件未满足而被暂停的过程就被称为等待，一个线程使得其他线程的保护条件得以满足的时候唤醒那些被暂停的线程的过程就被称为通知。

1.wait

  在Java平台中，Object.wait方法可以用来实现等待，下面是wait方法的三个重载方法：

• void wait(long timeoutMillis)
  调用该方法会使线程进入TIMED_WAITING状态，当等待时间结束或其他线程调用了该对象的notify或notifyAll方法时会将该线程唤醒。
• void wait(long timeoutMillis, int nanos)
  这个方法看上去可以精确到纳秒级别，但实际上并不是。如果nanos的值在0~999999之间，就给timeoutMillis加1，然后调用wait(timeoutMillis)。
• void wait()
  该方法相当于wait(0)，即永不超时。调用后当前线程会进入WAITING状态，直到其他线程调用了该对象的notify或notifyAll方法。

  先通过一张图来介绍wait的实现机制：

  在上一篇文章中我们了解到JVM会为每个对象维护一个入口集（Entry Set）用于存储申请该对象内部锁的线程。此外，JVM还会为每个对象维护一个被称为等待集（Wait Set）的队列，该队列用于存储该对象上的等待线程。当在线程中调用某个对象（这里我们称之为对象A）的wait方法后，当前线程会释放内部锁并进入WAITING或TIMED_WAITING状态，然后进入等待集中。当其他线程调用对象A的notify方法后，等待集中的某个线程会被唤醒并被移出等待集。这个线程可能会马上获得内部锁，也有可能因竞争内部锁失败而进入入口集，直到获得内部锁。当重新获取到内部锁后，wait方法才会返回，当前线程继续执行后面的代码。
  由于wait方法会释放内部锁，因此在wait方法中会判断当前线程是否持有被调用wait方法的对象的内部锁。如果当前线程没有持有该对象的内部锁，JVM会抛出一个IllegalMonitorStateException异常。因此，wait方法在调用时当前线程必须持有该对象的内部锁，即wait方法的调用必须要放在由该对象引导的synchronized同步块中。综上所述，使用wait方法实现等待的代码模板如下伪代码所示：

synchronized(someObject) {
    while(!someCondition) {
        someObject.wait();
    }
    doSomething();
}

  这里使用while而不是if的原因是，通知线程可能只是更新了保护条件中的共享变量，但并不一定会使保护条件成立；即使通知线程可以保证保护条件成立，但是在线程从等待集进入入口集再到获取到内部锁的这段时间内，其他线程仍然可能更新共享变量而导致保护条件不成立。线程虽然因为保护条件不成立而进入wait方法，但wait方法的返回并不能说明保护条件已经成立。因此，在wait方法返回后需要再次进行判断，若保护条件成立则执行接下来的操作，否则应该继续进入wait方法。正是基于这种考虑，我们应该将wait方法的调用放在while循环而不是if判断中。

2.notify/notifyAll

  下图是notify的实现机制：

  和wait方法一样，notify方法在执行时也必须持有对象的内部锁，否则会抛出IllegalMonitorStateException异常，因此notify方法也必须放在由该对象引导的synchronized同步块中。notify方法会将等待集中的任意一个线程移出队列。和wait方法不同的是，notify方法本身不会释放内部锁，而是在临界区代码执行完成后自动释放。因此，为了使等待线程在其被唤醒之后能够尽快获得内部锁，应该尽可能地将notify调用放在靠近临界区结束的地方。
  调用notify方法所唤醒的线程是相应对象上的一个任意等待线程，但是这个被唤醒的线程可能不是我们真正想要唤醒的那个线程。因此，有时候我们需要借助notifyAll，它和notify方法的唯一不同之处在于它可以唤醒相应对象上的所有等待线程。

3.过早唤醒问题

  假设通知线程N和等待线程W1和W2同步在对象obj上，W1和W2的保护条件C1和C2均依赖于obj的实例变量state，但C1和C2判断的内容并不相同。初始状态下C1和C2均不成立。某一时刻，当线程N更新了共享变量state使得保护条件C1得以成立，此时为了唤醒W1而执行了obj.notifyAll()方法（调用obj.notify()并不一定会唤醒W1）。由于notifyAll唤醒的是obj上的所有等待线程，因此W2也会被唤醒，即使W2的保护条件并未成立。这就使得W2在被唤醒之后仍然需要继续等待。这种等待线程在保护条件并未成立的情况下被唤醒的现象被称为过早唤醒。过早唤醒使得那些无需被唤醒的等待线程也被唤醒了，造成了资源的浪费。过早唤醒问题可以利用下一节中介绍的Condition接口来解决。

二.条件变量Condition

  总的来说，Object.wait()/notify()过于底层，且Object.wait(long timeout)还存在过早唤醒和无法区分其返回是由于等待超时还是被通知线程唤醒的问题。不过，了解wait/notify有助于我们阅读部分源码，以及学习和使用Condition接口。
  Condition接口可以作为wait/notify的替代品来实现等待/通知，它为解决过早唤醒问题提供了支持，并解决了Object.wait(long timeout)无法区分其返回是由于等待超时还是被通知线程唤醒的问题。Condition接口中定义了以下方法：

  在上一篇文章中，我们在介绍Lock接口时曾经提到过它的newCondition方法，它返回的就是一个Condition实例。类似于Object.wait()/notify()要求其执行线程必须持有这些方法所属对象的内部锁，Condition.await()/signal()也要求其执行线程持有创建该Condition实例的显式锁。每个Condition实例内部都维护了一个用于存储等待线程的队列。设condition1和condition2是从一个显式锁上获取的两个不同的Condition实例，一个线程执行condition1.await()会导致其被暂停并进入condition1的等待队列。condition1.signal()会使condition1的等待队列中的一个任意线程被唤醒，而condition1.signaAll()则会使condition1的等待队列中的所有线程被唤醒，而condition2的等待队列中的线程则不受影响。
  和wait/notify类似，await/signal的使用方法如下：

public class ConditionUsage {
    private final Lock lock = new ReentrantLock();
    private final Condition condition = lock.newCondition();

    public void waitMethod() throws InterruptedException {
        lock.lock();
        try {
            while (保护条件不成立) {
                condition.await();
            }
            // 业务逻辑
        } finally {
            lock.unlock();
        }
    }
    
    public void notifyMethod() {
        lock.unlock();
        try {
            // 更新共享变量
            condition.signal();
        } finally {
            lock.unlock();
        }
    }
}

  最后，以一个例子来结束本小节。这里我们以经典的生产者-消费者模型来举例。假设有一个生产整数的生产者，一个消费奇数的消费者和一个消费偶数的消费者。当生产奇数时，生产者会通知奇数消费者，偶数同理。下面是完整代码：

展开查看

import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class ConditionDemo {
    private final Lock lock = new ReentrantLock();
    private final Condition oddCondition = lock.newCondition();
    private final Condition evenCondition = lock.newCondition();
    private final Random random = new Random();
    private volatile Integer message;
    private AtomicInteger count = new AtomicInteger(0);

    public static void main(String[] args) {
        ConditionDemo demo = new ConditionDemo();
        Thread producer = new Thread(() -> {
            while (true) {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                demo.produce();
            }
        });
        producer.start();
        Thread oddConsumer = new Thread(() -> {
            while (true) {
                try {
                    demo.consumeOdd();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        Thread evenConsumer = new Thread(() -> {
            while (true) {
                try {
                    demo.consumeEven();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
        oddConsumer.start();
        evenConsumer.start();
    }

    public void produce() {
        lock.lock();
        if (message == null) {
            message = random.nextInt(100) + 1;
            count.incrementAndGet();
            if (message % 2 == 0) {
                evenCondition.signal();
                System.out.println("Produce even : " + message);
            } else {
                oddCondition.signal();
                System.out.println("Produce odd : " + message);
            }
        }
        lock.unlock();
    }

    public void consumeOdd() throws InterruptedException {
        lock.lock();
        while (message == null) {
            oddCondition.await();
        }
        System.out.println("Consume odd : " + message);
        message = null;
        lock.unlock();
    }

    public void consumeEven() throws InterruptedException {
        lock.lock();
        while (message == null) {
            evenCondition.await();
        }
        System.out.println("Consume even : " + message);
        message = null;
        lock.unlock();
    }
}

  该程序的输出如下：

Produce even : 34
Consume even : 34
Produce odd : 43
Consume odd : 43
Produce even : 28
Consume even : 28
Produce odd : 27
Consume odd : 27
Produce even : 92
Consume even : 92
...

三.倒数计数器CountDownLatch

  有时候，我们希望一个线程在另一个或多个线程结束之后再继续执行，这时候我们最先想到的肯定是Thread.join()。有时我们又希望一个线程不一定需要其他线程结束，而只是等其他线程执行完特定的操作就继续执行。这种情况下无法使用Thread.join()，因为它会导致当前线程等待其他线程完全结束。当然，此时可以用共享变量来实现。不过，Java为我们提供了更加方便的工具类来解决上面说的这些情况，那就是CountDownLatch。
  可以将CountDownLatch理解为一个可以在多个线程之间使用的计数器。这个类提供了以下方法：

  CountDownLatch内部也维护了一个用于存放等待线程的队列。当计数器不为0时，调用await方法的线程会被暂停并进入该队列。当某个线程调用countDown方法的时候，计数器会减1。当计数器到0的时候，等待队列中的所有线程都会被唤醒。计数器的初始值是在CountDownLatch的构造方法中指定的：

public CountDownLatch(int count)

  当计数器的值达到0之后就不会再变化。此时，调用countDown方法并不会导致异常的抛出，并且后续执行await方法的线程也不会被暂停。因此，CountDownLatch的使用是一次性的。此外，由于CountDownLatch是线程安全的，因此在调用await、countDown方法时无需加锁。
  下面的例子中，主线程等待两个子线程结束之后再继续执行。这里使用了CountDownLatch来实现：

import java.util.concurrent.CountDownLatch;

public class CountDownLatchDemo {
    public static void main(String[] args) {
        CountDownLatch latch = new CountDownLatch(2);
        Runnable task = () -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(Thread.currentThread().getName() + " finished.");
            latch.countDown();
        };
        new Thread(task, "Thread 1").start();
        new Thread(task, "Thread 2").start();
        try {
            latch.await();
        } catch (InterruptedException e) {
            return;
        }
        System.out.println("Main thread continued.");
    }
}

  该程序输出如下：

Thread 2 finished.
Thread 1 finished.
Main thread continued.

  可以看到，当线程1和线程2执行完成后，主线程才开始继续执行。
  如果CountDownLatch内部计数器由于程序的错误而永远无法达到0，那么相应实例上的等待线程会一直处于WAITING状态。避免该问题的出现有两种方法：一是确保所有对countDown方法的调用都位于代码中正确的位置，例如放在finally块中。二是使用带有时间限制的await方法。如果在规定时间内计时器值未达到0，该CountDownLatch实例上的等待线程也会被唤醒。该方法的返回值可以用于区分其返回是否是由于等待超时。
  此外，对于同一个CountDownLatch实例latch，latch.countDown()的执行线程在执行该方法之前所执行的任何内存操作对等待线程在latch.await()调用之后的代码是可见的且有序的。

四.循环屏障CyclicBarrier

  有时候多个线程可能需要互相等待对方执行到代码中的某个地方才能继续执行。这就类似于我们在开会的时候必须等待所有与会人员都到场之后才能开始。Java中为我们提供了一个工具类CyclicBarrier，该类可以用来实现这种等待。
  使用CyclicBarrier实现等待的线程被称为参与方（Party）。参与方只需要CyclicBarrier.await()就可以实现等待。和CountDownLatch类似，CyclicBarrier也有一个计数器。当最后一个线程调用CyclicBarrier.await()时，之前的等待线程都会被唤醒，而最后一个线程本身并不会被暂停。和CountDownLatch不同的是，CyclicBarrier是可以重复使用的，这也是为什么它的类名中含有Cyclic。当所有参与方被唤醒的时候，任何线程再次执行await方法又会导致该线程被暂停。
  CyclicBarrier提供了两个构造器：

public CyclicBarrier​(int parties)
public CyclicBarrier​(int parties, Runnable barrierAction)

  可以看到，在构造CyclicBarrier​时，必须提供参与方的数量。第二个构造器还允许我们指定一个被称为barrierAction的任务（Runnable接口实例），该任务会被最后一个执行await方法的线程执行。因此，如果有需要在唤醒所有线程前执行的操作，可以使用这个构造器。
  CyclicBarrier提供了以下6个方法：
1.public int await() throws InterruptedException,BrokenBarrierException
  如果当前线程不是最后一个参与方，那么该线程在调用await()后将持续等待直到以下情况发生：

• 最后一个线程到达；
• 当前线程被中断；
• 其他正在等待的线程被中断；
• 其他线程等待超时；
• 其他线程调用了当前屏障的reset()。

  如果当前线程在进入await()方法使已经被标记中断状态或在等待时被中断，那么await()将会抛出InterruptedException并清除当前线程的中断状态。
  如果屏障在参与方等待时被重置或被破坏，或者在调用await()时屏障已经被破坏，那么await()将会抛出BrokenBarrierException。
  如果某个线程在等待时被中断，那么其他等待线程将会抛出BrokenBarrierException并且屏障也会被标记为broken状态。
  该方法的返回值表示当前线程的到达索引，getParties()-1表示第一个到达，0表示最后一个到达。
2.public int await​(long timeout,TimeUnit unit) throws InterruptedException,BrokenBarrierException,TimeoutException
  该方法与相当于有时间限制的await()，等待时间结束之后该线程将会抛出TimeOutException，屏障会被标记为broken状态，其他正在等待的线程则会抛出BrokenBarrierException。
3.public int getNumberWaiting()
  返回当前正在等待的参与方的数量。
4.public int getParties()
  返回总的参与方的数量。
5.public boolean isBroken()
  如果该屏障已经被破坏则返回true，否则返回false。当等待线程超时或被中断，或者在执行barrierAction时出现异常，屏障将会被破坏。
6.public void reset()
  将屏障恢复到初始状态，如果有正在等待的线程，这些线程会抛出BrokenBarrierException异常。

  下面我们通过一个例子来学习如何使用CyclicBarrier。假设现在正在举行短跑比赛，共有8名参赛选手，而场地上只有4条赛道，因此需要分为两场比赛。每场比赛必须等4名选手全都就绪才可以开始，而上一场比赛结束之后即全部选手离开赛道之后才能进行下一场比赛。该示例代码如下所示：

展开查看

import java.util.Random;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.atomic.AtomicInteger;

public class CyclicBarrierDemo {
    private CyclicBarrier startBarrier = new CyclicBarrier(4, () -> System.out.println("比赛开始！"));
    private CyclicBarrier shiftBarrier = new CyclicBarrier(4, () -> System.out.println("比赛结束！"));
    private Runner[] runners = new Runner[8];
    private AtomicInteger next = new AtomicInteger(0);

    CyclicBarrierDemo() {
        for (int i = 0; i < 8; i++) {
            runners[i] = new Runner(i / 4 + 1, i % 4 + 1);
        }
    }

    public static void main(String[] args) {
        CyclicBarrierDemo demo = new CyclicBarrierDemo();
        for (int i = 0; i < 4; i++) {
            demo.new Track().start();
        }
    }

    private class Track extends Thread {
        private Random random = new Random();

        @Override
        public void run() {
            for (int i = 0; i < 2; i++) {
                try {
                    Runner runner = runners[next.getAndIncrement()];
                    System.out.println(runner.getGroup() + "组" + runner.getNumber() + "号准备就绪！");
                    startBarrier.await();
                    System.out.println(runner.getGroup() + "组" + runner.getNumber() + "号出发！");
                    Thread.sleep((random.nextInt(5) + 1) * 1000);
                    System.out.println(runner.getGroup() + "组" + runner.getNumber() + "号到达终点！");
                    shiftBarrier.await();
                } catch (InterruptedException | BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    private static class Runner {
        private int group;
        private int number;

        Runner(int group, int number) {
            this.group = group;
            this.number = number;
        }

        int getGroup() {
            return group;
        }

        int getNumber() {
            return number;
        }
    }
}

  该程序输出如下：

展开查看

1组4号准备就绪！
1组2号准备就绪！
1组3号准备就绪！
1组1号准备就绪！
比赛开始！
1组4号出发！
1组2号出发！
1组1号出发！
1组3号出发！
1组3号到达终点！
1组2号到达终点！
1组4号到达终点！
1组1号到达终点！
比赛结束！
2组1号准备就绪！
2组2号准备就绪！
2组3号准备就绪！
2组4号准备就绪！
比赛开始！
2组4号出发！
2组1号出发！
2组3号出发！
2组2号出发！
2组1号到达终点！
2组4号到达终点！
2组3号到达终点！
2组2号到达终点！
比赛结束！

五.总结

  等待线程可以通过执行Object.wait()/wait(long)来实现等待，通知线程可以通过执行Object.notify()/notifyAll()来实现通知。等待线程和通知线程在执行Object.wait()/wait(long)、Object.notify()/notifyAll()时必须持有相应对象对应的内部锁。为了保证线程被唤醒时保护条件一定是成立的，应该将对保护条件的判断、Object.wait()/wait(long)的调用放在相应对象所引导的临界区中的一个循环之中。
  条件变量（Condition接口）是wait/notify的替代品。Condition接口对解决过早唤醒问题提供了支持，它的await(long)还解决了Object.wait(long)无法区分其返回是否是因为等待超时的问题。
  CountDownLatch能够用来实现一个线程等待其他线程执行的特定操作的结束。等待线程执行CountDownLatch.await()，通知线程执行CountDownLatch.countDown()。为避免等待线程永远处于暂停状态而无法被唤醒，对countDown()的调用通常需要被放在finally块中。一个CountDownLatch实例只能实现一次等待/通知。对于同一个CountDownLatch实例latch，latch.countDown()的执行线程在执行该方法之前所执行的任何内存操作对等待线程在latch.await()调用之后的代码是可见的且有序的。
  CyclicBarrier能够用于实现多个线程间的相互等待。CyclicBarrier.await()既是等待方法又是通知方法。CyclicBarrier实例的所有参与方除最后一个线程外都相当于等待线程，最后一个线程则相当于通知线程。与CountDownLatch 不同的是，CyclicBarrier实例是可以复用的，一个CountDownLatch实例可以实现多次等待/通知。
  线程间的通信方式远不止上面介绍的这些，在下一篇文章中，我们将会继续学习线程间的通信方式。