Java concurrency之Condition条件_动力节点Java学院整理

condition介绍

condition的作用是对锁进行更精确的控制。condition中的await()方法相当于object的wait()方法，condition中的signal()方法相当于object的notify()方法，condition中的signalall()相当于object的notifyall()方法。不同的是，object中的wait(),notify(),notifyall()方法是和"同步锁"(synchronized关键字)捆绑使用的；而condition是需要与"互斥锁"/"共享锁"捆绑使用的。

condition函数列表

// 造成当前线程在接到信号或被中断之前一直处于等待状态。
void await()
// 造成当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态。
boolean await(long time, timeunit unit)
// 造成当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态。
long awaitnanos(long nanostimeout)
// 造成当前线程在接到信号之前一直处于等待状态。
void awaituninterruptibly()
// 造成当前线程在接到信号、被中断或到达指定最后期限之前一直处于等待状态。
boolean awaituntil(date deadline)
// 唤醒一个等待线程。
void signal()
// 唤醒所有等待线程。
void signalall()

condition示例

示例1是通过object的wait(), notify()来演示线程的休眠/唤醒功能。

示例2是通过condition的await(), signal()来演示线程的休眠/唤醒功能。

示例3是通过condition的高级功能。

示例1

public class waittest1 {
  public static void main(string[] args) {
    threada ta = new threada("ta");
    synchronized(ta) { // 通过synchronized(ta)获取“对象ta的同步锁”
      try {
        system.out.println(thread.currentthread().getname()+" start ta");
        ta.start();
        system.out.println(thread.currentthread().getname()+" block");
        ta.wait();  // 等待
        system.out.println(thread.currentthread().getname()+" continue");
      } catch (interruptedexception e) {
        e.printstacktrace();
      }
    }
  }
  static class threada extends thread{
    public threada(string name) {
      super(name);
    }
    public void run() {
      synchronized (this) { // 通过synchronized(this)获取“当前对象的同步锁”
        system.out.println(thread.currentthread().getname()+" wakup others");
        notify();  // 唤醒“当前对象上的等待线程”
      }
    }
  }
}

示例2

import java.util.concurrent.locks.lock;
import java.util.concurrent.locks.condition;
import java.util.concurrent.locks.reentrantlock;
public class conditiontest1 {
  private static lock lock = new reentrantlock();
  private static condition condition = lock.newcondition();
  public static void main(string[] args) {
    threada ta = new threada("ta");
    lock.lock(); // 获取锁
    try {
      system.out.println(thread.currentthread().getname()+" start ta");
      ta.start();
      system.out.println(thread.currentthread().getname()+" block");
      condition.await();  // 等待
      system.out.println(thread.currentthread().getname()+" continue");
    } catch (interruptedexception e) {
      e.printstacktrace();
    } finally {
      lock.unlock();  // 释放锁
    }
  }
  static class threada extends thread{
    public threada(string name) {
      super(name);
    }
    public void run() {
      lock.lock();  // 获取锁
      try {
        system.out.println(thread.currentthread().getname()+" wakup others");
        condition.signal();  // 唤醒“condition所在锁上的其它线程”
      } finally {
        lock.unlock();  // 释放锁
      }
    }
  }
}

运行结果：

main start ta
main block
ta wakup others
main continue

通过“示例1”和“示例2”，我们知道condition和object的方法有一下对应关系：

              object      condition 
休眠          wait        await
唤醒个线程     notify      signal
唤醒所有线程   notifyall   signalall

condition除了支持上面的功能之外，它更强大的地方在于：能够更加精细的控制多线程的休眠与唤醒。对于同一个锁，我们可以创建多个condition，在不同的情况下使用不同的condition。

例如，假如多线程读/写同一个缓冲区：当向缓冲区中写入数据之后，唤醒"读线程"；当从缓冲区读出数据之后，唤醒"写线程"；并且当缓冲区满的时候，"写线程"需要等待；当缓冲区为空时，"读线程"需要等待。         如果采用object类中的wait(), notify(), notifyall()实现该缓冲区，当向缓冲区写入数据之后需要唤醒"读线程"时，不可能通过notify()或notifyall()明确的指定唤醒"读线程"，而只能通过notifyall唤醒所有线程(但是notifyall无法区分唤醒的线程是读线程，还是写线程)。  但是，通过condition，就能明确的指定唤醒读线程。

看看下面的示例3，可能对这个概念有更深刻的理解。 

示例3

import java.util.concurrent.locks.lock;
import java.util.concurrent.locks.condition;
import java.util.concurrent.locks.reentrantlock;
class boundedbuffer {
  final lock lock = new reentrantlock();
  final condition notfull = lock.newcondition(); 
  final condition notempty = lock.newcondition(); 
  final object[] items = new object[5];
  int putptr, takeptr, count;
  public void put(object x) throws interruptedexception {
    lock.lock();  //获取锁
    try {
      // 如果“缓冲已满”，则等待；直到“缓冲”不是满的，才将x添加到缓冲中。
      while (count == items.length)
        notfull.await();
      // 将x添加到缓冲中
      items[putptr] = x; 
      // 将“put统计数putptr+1”；如果“缓冲已满”，则设putptr为0。
      if (++putptr == items.length) putptr = 0;
      // 将“缓冲”数量+1
      ++count;
      // 唤醒take线程，因为take线程通过notempty.await()等待
      notempty.signal();
      // 打印写入的数据
      system.out.println(thread.currentthread().getname() + " put "+ (integer)x);
    } finally {
      lock.unlock();  // 释放锁
    }
  }
  public object take() throws interruptedexception {
    lock.lock();  //获取锁
    try {
      // 如果“缓冲为空”，则等待；直到“缓冲”不为空，才将x从缓冲中取出。
      while (count == 0) 
        notempty.await();
      // 将x从缓冲中取出
      object x = items[takeptr]; 
      // 将“take统计数takeptr+1”；如果“缓冲为空”，则设takeptr为0。
      if (++takeptr == items.length) takeptr = 0;
      // 将“缓冲”数量-1
      --count;
      // 唤醒put线程，因为put线程通过notfull.await()等待
      notfull.signal();
      // 打印取出的数据
      system.out.println(thread.currentthread().getname() + " take "+ (integer)x);
      return x;
    } finally {
      lock.unlock();  // 释放锁
    }
  } 
}
public class conditiontest2 {
  private static boundedbuffer bb = new boundedbuffer();
  public static void main(string[] args) {
    // 启动10个“写线程”，向boundedbuffer中不断的写数据(写入0-9)；
    // 启动10个“读线程”，从boundedbuffer中不断的读数据。
    for (int i=0; i<10; i++) {
      new putthread("p"+i, i).start();
      new takethread("t"+i).start();
    }
  }
  static class putthread extends thread {
    private int num;
    public putthread(string name, int num) {
      super(name);
      this.num = num;
    }
    public void run() {
      try {
        thread.sleep(1);  // 线程休眠1ms
        bb.put(num);    // 向boundedbuffer中写入数据
      } catch (interruptedexception e) {
      }
    }
  }
  static class takethread extends thread {
    public takethread(string name) {
      super(name);
    }
    public void run() {
      try {
        thread.sleep(10);          // 线程休眠1ms
        integer num = (integer)bb.take();  // 从boundedbuffer中取出数据
      } catch (interruptedexception e) {
      }
    }
  }
}

(某一次)运行结果：

p1 put  1
p4 put  4
p5 put  5
p0 put  0
p2 put  2
t0 take 1
p3 put  3
t1 take 4
p6 put  6
t2 take 5
p7 put  7
t3 take 0
p8 put  8
t4 take 2
p9 put  9
t5 take 3
t6 take 6
t7 take 7
t8 take 8
t9 take 9

结果说明：

(01) boundedbuffer 是容量为5的缓冲，缓冲中存储的是object对象，支持多线程的读/写缓冲。多个线程操作“一个boundedbuffer对象”时，它们通过互斥锁lock对缓冲区items进行互斥访问；而且同一个boundedbuffer对象下的全部线程共用“notfull”和“notempty”这两个condition。

       notfull用于控制写缓冲，notempty用于控制读缓冲。当缓冲已满的时候，调用put的线程会执行notfull.await()进行等待；当缓冲区不是满的状态时，就将对象添加到缓冲区并将缓冲区的容量count+1，最后，调用notempty.signal()缓冲notempty上的等待线程(调用notempty.await的线程)。 简言之，notfull控制“缓冲区的写入”，当往缓冲区写入数据之后会唤醒notempty上的等待线程。

       同理，notempty控制“缓冲区的读取”，当读取了缓冲区数据之后会唤醒notfull上的等待线程。

(02) 在conditiontest2的main函数中，启动10个“写线程”，向boundedbuffer中不断的写数据(写入0-9)；同时，也启动10个“读线程”，从boundedbuffer中不断的读数据。

(03) 简单分析一下运行结果。

     1, p1线程向缓冲中写入1。    此时，缓冲区数据:   | 1 |   |   |   |   |

     2, p4线程向缓冲中写入4。    此时，缓冲区数据:   | 1 | 4 |   |   |   |

     3, p5线程向缓冲中写入5。    此时，缓冲区数据:   | 1 | 4 | 5 |   |   |

     4, p0线程向缓冲中写入0。    此时，缓冲区数据:   | 1 | 4 | 5 | 0 |   |

     5, p2线程向缓冲中写入2。    此时，缓冲区数据:   | 1 | 4 | 5 | 0 | 2 |

     此时，缓冲区容量为5；缓冲区已满！如果此时，还有“写线程”想往缓冲中写入数据，会调用put中的notfull.await()等待，直接缓冲区非满状态，才能继续运行。

     6, t0线程从缓冲中取出数据1。此时，缓冲区数据:   |   | 4 | 5 | 0 | 2 |

     7, p3线程向缓冲中写入3。    此时，缓冲区数据:   | 3 | 4 | 5 | 0 | 2 |

     8, t1线程从缓冲中取出数据4。此时，缓冲区数据:   | 3 |   | 5 | 0 | 2 |

     9, p6线程向缓冲中写入6。    此时，缓冲区数据:   | 3 | 6 | 5 | 0 | 2 |
     ...