Java 线程通信深入理解Java 线程通信

时间:2021-04-22 认真对待世界的小白人气:0

当线程在系统内运行时，线程的调度具有一定的透明性，程序通常无法准确控制线程的轮换执行，但 Java 也提供了一些机制来保证线程协调运行。

传统的线程通信

假设现在系统中有两个线程，这两个线程分别代表存款者和取钱者——现在假设系统有一种特殊的要求，系统要求存款者和取钱者不断地重复存款、取钱的动作，而且要求每当存款者将钱存入指定账户后，取钱者就立即取出该笔钱。不允许存款者连续两次存钱，也不允许取钱者连续两次取钱。

为了实现这种功能，可以借助于 Object 类提供的 wait()、 notify() 和 notifyAll() 三个方法，这三个方法并不属于 Thread 类，而是属于 Object 类。但这三个方法必须由同步监视器对象来调用，这可分成以下两种情况。

对于使用 synchronized 修饰的同步方法，因为该类的默认实例（this）就是同步监视器，所以可以在同步方法中直接调用这三个方法。
对于使用 synchronized 修饰的同步代码块，同步监视器是 synchronized 后括号里的对象，所以必须使用该对象调用这三个方法。

关于这三个方法的解释如下。

wait()：导致当前线程等待，直到其他线程调用该同步监视器的 notify() 方法或 notifyAll() 方法来唤醒该线程。该 wait() 方法有三种形式——无时间参数的 wait (—直等待，直到其他线程通知）、带毫秒参数的 wait() 和带毫秒、毫微秒参数的 wait() （这两种方法都是等待指定时间后自动苏醒）。调用 wait() 方法的当前线程会释放对该同步监视器的锁定。
notify()：唤醒在此同步监视器上等待的单个线程。如果所有线程都在此同步监视器上等待，则会选择唤醒其中一个线程。选择是任意性的。只有当前线程放弃对该同步监视器的锁定后（使用 wait() 方法），才可以执行被唤醒的线程。
notifyAll()：唤醒在此同步监视器上等待的所有线程。只有当前线程放弃对该同步监视器的锁定后，才可以执行被唤醒的线程。

程序中可以通过一个旗标来标识账户中是否已有存款，当旗标为 false 时，表明账户中没有存款，存款者线程可以向下执行，当存款者把钱存入账户后，将旗标设为 true ，并调用 notify() 或 notifyAll() 方法来唤醒其他线程；当存款者线程进入线程体后，如果旗标为 true 就调用 wait() 方法让该线程等待。

当旗标为 true 时，表明账户中已经存入了存款，则取钱者线程可以向下执行，当取钱者把钱从账户中取出后，将旗标设为 false ，并调用 notify() 或 notifyAll() 方法来唤醒其他线程；当取钱者线程进入线程体后，如果旗标为 false 就调用 wait() 方法让该线程等待。

本程序为 Account 类提供 draw() 和 deposit() 两个方法，分别对应该账户的取钱、存款等操作，因为这两个方法可能需要并发修改 Account 类的 balance 成员变量的值，所以这两个方法都使用 synchronized 修饰成同步方法。除此之外，这两个方法还使用了 wait() 和 notifyAll() 来控制线程的协作。

public class Account{
  private String accountNo;
  private double balance;
  //标识账户中是否已有存款的旗标
  private boolean flag = false;

  public Account(){}

  public Account(String accountNo , double balance){
    this.accountNo = accountNo;
    this.balance = balance;
  }

  public void setAccountNo(String accountNo){
    this.accountNo = accountNo;
  }
  public String getAccountNo(){
     return this.accountNo;
  }

  public double getBalance(){
     return this.balance;
  }
  public synchronized void draw(double drawAmount){
    try{
      //如果flag为假，表明账户中还没有人存钱进去，则取钱方法阻塞
      if (!flag){
        wait();
      }else{
        //执行取钱
        System.out.println(Thread.currentThread().getName() + " 取钱:" + drawAmount);
        balance -= drawAmount;
        System.out.println("账户余额为：" + balance);
        //将标识账户是否已有存款的旗标设为false。
        flag = false;
        //唤醒其他线程
        notifyAll();
      }
    }catch (InterruptedException ex){
      ex.printStackTrace();
    }
  }
  public synchronized void deposit(double depositAmount){
    try{
      //如果flag为真，表明账户中已有人存钱进去，则存钱方法阻塞
      if (flag){  // ①
        wait();
      }else{
        //执行存款
        System.out.println(Thread.currentThread().getName() + " 存款:" + depositAmount);
        balance += depositAmount;
        System.out.println("账户余额为：" + balance);
        //将表示账户是否已有存款的旗标设为true
        flag = true;
        //唤醒其他线程
        notifyAll();
      }
    }catch (InterruptedException ex){
      ex.printStackTrace();
    }
  }

  public int hashCode(){
    return accountNo.hashCode();
  }
  public boolean equals(Object obj){
    if (obj != null && obj.getClass() == Account.class){
      Account target = (Account)obj;
      return target.getAccountNo().equals(accountNo);
    }
    return false;
  }
}

上面程序中的粗体字代码使用 wait() 和 notifyAll() 进行了控制，对存款者线程而言，当程序进入 deposit() 方法后，如果 flag 为 true ，则表明账户中已有存款，程序调用 wait() 方法阻塞；否则程序向下执行存款操作，当存款操作执行完成后，系统将 flag 设为 true，然后调用 notifyAll() 来唤醒其他被阻塞的线程——如果系统中有存款者线程，存款者线程也会被唤醒，但该存款者线程执行到①号代码处时再次进入阻塞状态，只有执行 draw() 方法的取钱者线程才可以向下执行。同理，取钱者线程的运行流程也是如此。

程序中的存款者线程循环100次重复存款，而取钱者线程则循环100次重复取钱，存款者线程和取钱者线程分别调用 Account 对象的 deposit()、 draw() 方法来实现。

public class DrawThread extends Thread{
  //模拟用户账户
  private Account account;
  //当前取钱线程所希望取的钱数
  private double drawAmount;

  public DrawThread(String name, Account account, double drawAmount){
    super(name);
    this.account = account;
    this.drawAmount = drawAmount;
  }

  //重复100次执行取钱操作
  public void run(){
    for (int i = 0 ; i < 100 ; i++ ){
      account.draw(drawAmount);
    }
  }
}

public class DepositThread extends Thread{
  //模拟用户账户
  private Account account;
  //当前取钱线程所希望存款的钱数
  private double depositAmount;

  public DepositThread(String name, Account account, double depositAmount){
    super(name);
    this.account = account;
    this.depositAmount = depositAmount;
  }

  //重复100次执行存款操作
  public void run(){
    for (int i = 0 ; i < 100 ; i++ ){
      account.deposit(depositAmount);
    }    
  }
}

主程序可以启动任意多个存款线程和取钱线程，可以看到所有的取钱线程必须等存款线程存钱后才可以向下执行，而存款线程也必须等取钱线程取钱后才可以向下执行。主程序代码如下。

public class TestDraw{
  public static void main(String[] args){
    //创建一个账户
    Account acct = new Account("1234567" , 0);
    new DrawThread("取钱者" , acct , 800).start();
    new DepositThread("存款者甲" , acct , 800).start();
    new DepositThread("存款者乙" , acct , 800).start();
    new DepositThread("存款者丙" , acct , 800).start();
  }
}

运行该程序，可以看到存款者线程、取钱者线程交替执行的情形，每当存款者向账户中存入800元之后，取钱者线程立即从账户中取出这笔钱。存款完成后账户余额总是800元，取钱结束后账户余额总是0元。运行该程序，会看到如下图所示的结果。

从上图中可以看出， 3个存款者线程随机地向账户中存款，只有1个取钱者线程执行取钱操作。只有当取钱者取钱后，存款者才可以存款；同理，只有等存款者存款后，取钱者线程才可以取钱。

上图显示程序最后被阻塞无法继续向下执行，这因为3个存款者线程共有300次存款操作，但1个取钱者线程只有100次取钱操作，所以程序最后被阻塞。

注意：上图所示的阻塞并不是死锁，对于这种情况，取钱者线程已经执行结束，而存款者线程只是在等待其他线程来取钱而已，并不是等待其他线程释放同步监视器。不要把死锁和程序阻塞等同起来！

使用 Condition 控制线程通信

如果程序不使用 synchronized 关键字来保证同步，而是直接使用 Lock 对象来保证同步，则系统中不存在隐式的同步监视器，也就不能使用 wait()、notify()、notifyAll() 方法进行线程通信了。

当使用 Lock 对象来保证同步时，Java 提供了一个 Condition 类来保持协调，使用 Condition 可以让那些已经得到 Lock 对象却无法继续执行的线程释放 Lock 对象，Condition 对象也可以唤醒其他处于等待的线程。

Condition 将同步监视器方法（wait()、notify() 和 notifyAll() ）分解成截然不同的对象，以便通过将这些对象与 Lock 对象组合使用，为每个对象提供多个等待集（wait-set）。在这种情况下，Lock 替代了同步方法或同步代码块，Condition 替代了同步监视器的功能。

Condition 实例被绑定在一个 Lock 对象上。要获得特定 Lock 实例的 Condition 实例，调用 Lock 对象的 newCondition() 方法即可。Condition 类提供了如下三个方法。

await()：类似于隐式同步监视器上的 wait() 方法，导致当前线程等待，直到其他线程调用该 Condition 的 signal() 方法或 signalAll() 方法来唤醒该线程。该 await() 方法有更多变体，如 long awaitNanos(long nanosTimeout)、 void awaitUninterruptibly() 、 awaitUntil(Date deadline) 等，可以完成更丰富的等待操作。
signal()：唤醒在此 Lock 对象上等待的单个线程。如果所有线程都在该 Lock 对象上等待，则会选择唤醒其中一个线程。选择是任意性的。只有当前线程放弃对该 Lock 对象的锁定后（使用 await() 方法），才可以执行被唤醒的线程。
signalAll()：唤醒在此 Lock 对象上等待的所有线程。只有当前线程放弃对该 Lock 对象的锁定后，才可以执行被唤醒的线程。

下面程序中 Account 使用 Lock 对象来控制同步，并使用 Condition 对象来控制线程的协调运行。

public class Account{
  //显示定义Lock对象
  private final Lock lock = new ReentrantLock();
  //获得指定Lock对象对应的条件变量
  private final Condition cond = lock.newCondition(); 

  private String accountNo;
  private double balance;

  //标识账户中是否已经存款的旗标
  private boolean flag = false;

  public Account(){}

  public Account(String accountNo , double balance){
    this.accountNo = accountNo;
    this.balance = balance;
  }

  public void setAccountNo(String accountNo){
    this.accountNo = accountNo;
  }
  public String getAccountNo(){
     return this.accountNo;
  }

  public double getBalance(){
     return this.balance;
  }
  public void draw(double drawAmount){
    //加锁
    lock.lock();
    try{
      //如果账户中还没有存入存款，该线程等待
      if (!flag){
        cond.await();
      }else{
        //执行取钱操作
        System.out.println(Thread.currentThread().getName() + " 取钱:" + drawAmount);
        balance -= drawAmount;
        System.out.println("账户余额为：" + balance);
        //将标识是否成功存入存款的旗标设为false
        flag = false;
        //唤醒该Lock对象对应的其他线程
        cond.signalAll();
      }
    }catch (InterruptedException ex){
      ex.printStackTrace();
    }
    //使用finally块来确保释放锁
    finally{
      lock.unlock();
    }
  }
  public void deposit(double depositAmount){
    lock.lock();
    try{
      //如果账户中已经存入了存款，该线程等待
      if(flag){
        cond.await();        
      }else{
        //执行存款操作
        System.out.println(Thread.currentThread().getName() + " 存款:" + depositAmount);
        balance += depositAmount;
        System.out.println("账户余额为：" + balance);
        //将标识是否成功存入存款的旗标设为true
        flag = true;
        //唤醒该Lock对象对应的其他线程
        cond.signalAll();
      }
    }catch (InterruptedException ex){
      ex.printStackTrace();
    }
    //使用finally块来确保释放锁
    finally{
      lock.unlock();
    }
  }

  public int hashCode(){
    return accountNo.hashCode();
  }
  public boolean equals(Object obj){
    if (obj != null && obj.getClass() == Account.class){
      Account target = (Account)obj;
      return target.getAccountNo().equals(accountNo);
    }
    return false;
  }
}

显式地使用 Lock 对象来充当同步监视器，则需要使用 Condition 对象来暂停、唤醒指定线程。存取钱的代码和最上面相同。

使用阻塞队列（BlockingQueue）控制线程通信

Java5 提供了一个 BlockingQueue 接口，虽然 BlockingQueue 也是 Queue 的子接口，但它的主要用途并不是作为容器，而是作为线程同步的工具。 BlockingQueue 具有一个特征：当生产者线程试图向 BlockingQueue 中放入元素时，如果该队列已满，则该线程被阻塞；当消费者线程试图从 BlockingQueue 中取出元素时，如果该队列已空，则该线程被阻塞。

程序的两个线程通过交替向 BlockingQueue 中放入元素、取出元素，即可很好地控制线程的通信。BlockingQueue 提供如下两个支持阻塞的方法。

put(E e)：尝试把 E 元素放入 BlockingQueue 中，如果该队列的元素己满，则阻塞该线程。
take()：尝试从 BlockingQueue 的头部取出元素，如果该队列的元素已空，则阻塞该线程。

BlockingQueue 继承了 Queue 接口，当然也可使用 Queue 接口中的方法。这些方法归纳起来可分为如下三组。

在队列尾部插入元素。包括 add(E e)、offer(E e) 和 put(E e) 方法，当该队列已满时，这三个方法分别会抛出异常、返回 false 、阻塞队列。
在队列头部删除并返回删除的元素。包括 remove()、 poll() 和 take() 方法。当该队列已空时，这三个方法分别会抛出异常、返回 false 、阻塞队列。
在队列头部取出但不删除元素。包括 element() 和 peek() 方法，当队列已空时，这两个方法分别抛出异常、返回 false

BlockingQueue 包含的方法之间的对应关系如下表所示：

BlockingQueue 与其实现类之间的类图如下图所示。

上图中以黑色方框框出的都是 Java7 新增的阻塞队列。可以看到， BlockingQueue 包含如下5个实现类。

ArrayBlockingQueue：基于数组实现的 BlockingQueue 队列。
LinkedBlockingQueue：基于链表实现的 BlockingQueue 队列。
PriorityBlockingQueue：它并不是标准的阻塞队列。与前面介绍的 PriorityQueue 类似，该队列调用 remove()、poll()、take() 等方法取出元素时，并不是取出队列中存在时间最长的元素，而是队列中最小的元素。 PriorityBlockingQueue 判断元素的大小即可根据元素（实现 Comparable 接口）的本身大小来自然排序，也可使用 Comparator 进行定制排序。
SynchronousQueue：同步队列。对该队列的存、取操作必须交替进行。
DelayQueue：它是一个特殊的 BlockingQueue ，底层基于 PriorityBlockingQueue 实现。不过，DelayQueue 要求集合元素都实现 Delay 接口（该接口里只有一个 long getDelay() 方法），DelayQueue 根据集合元素的 getDalay() 方法的返回值进行排序。

下面以 ArrayBlockingQueue 为例介绍阻塞队列的功能和用法。下面先用一个最简单的程序来测试 BlockingQueue 的 put() 方法。

public class BlockingQueueTest {
  public static void main(String[] args) throws Exception {
    BlockingQueue<String> bq = new ArrayBlockingQueue<>(2);
    bq.put("Java"); // 与bq.add("Java")、bq.offer("Java") 相同
    bq.put("Java"); // 与bq.add("Java")、bq.offer("Java") 相同
    bq.put("Java"); // ① 阻塞线程
  }
}

上面程序先定义一个大小为2的 BlockingQueue，程序先向该队列中放入两个元素，此时队列还没有满，两个元紊都可以放入，因此使用 put()、add() 和 offer() 方法效果完全一样。当程序试图放入第三个元素时，如果使用 put() 方法尝试放入元素将会阻寒线程，如上面程序①号代码所示。如果使用 add() 方法尝试放入元素将会引发异常；如果使用 offer() 方法尝试放入元素则会返回 false，元素不会被放入。

与此类似的是，在 BlockingQueue 已空的情况下，程序使用 take() 方法尝试取出元素将会阻塞线程：使用 remove() 方法尝试取出元素将引发异常：使用 poll() 方法尝试取出元素将返回 false，元索不会被删除。

掌握了 BlodcingQuene 阻塞队列的特性之后，下面程序就可以利用 BlockingQueue 来实现线程通信了。

public class Producer extends Thread {

  private BlockingQueue<String> bq;
  public Producer(BlockingQueue<String> bq) {
    this.bq = bq;
  }
  public void run() {
    String[] strArr = new String[] {
        "Java",
        "Struts",
        "Spring"
    };
    
    for(int i=0;i<99999999;i++) {
      System.out.println(getName()+"生产者准备生产集合元素");
      try {
        Thread.sleep(200);
        // 尝试放入元素，如果队列已满，则线程被阻塞
        bq.put(strArr[i%3]);
      }catch(Exception ex) {
        ex.printStackTrace();
      }
      System.out.println(getName()+"生产完成："+bq);
    }
  }
}

public class Consumer extends Thread {
  private BlockingQueue<String> bq;
  public Consumer(BlockingQueue<String> bq) {
    this.bq = bq;
  }
  public void run() {
    while(true) {
      System.out.println(getName()+"消费者准备消费集合元素！");
      try {
        Thread.sleep(200);
        // 尝试取出元素，如果队列已空，则线程被阻塞
        bq.take();
      }catch(Exception ex) {
        ex.printStackTrace();
      }
      System.out.println(getName()+"消费完成："+bq);
    }
  }
}


public class BlockingQueueTest2 {
  public static void main(String[] args) {
    // 创建一个容量为1的BlockingQueue
    BlockingQueue<String> bq = new ArrayBlockingQueue<>(1);
    // 启动3个生产者线程
    new Producer(bq).start();
    new Producer(bq).start();
    new Producer(bq).start();
    // 启动一个消费者线程
    new Consumer(bq).start();
  }
}

上面程序启动了 3个生产者线程向 BlockingQueue 集合放入元素，启动了 1个消费者线程从 BlockingQueue 集合取出元素。本程序的 BlockingQueue 集合容量为1，因此3个生产者线程无法连续放入元素，必须等待消费者线程取出一个元素后， 3个生产者线程的其中之一才能放入一个元素。运行该程序，会看到如下图所示的结果。

从上图可以看出，3个生产者线程都想向 BlockingQueue 中放入元素，但只要其中一个线程向该队列中放入元素之后，其他生产者线程就必须等待，等待消费者线程取出 BlockingQueue 队列里的元素。

加载全部内容

Java 线程通信 深入理解Java 线程通信

Java 线程通信深入理解Java 线程通信