深入理解Java 线程通信
作者：认真对待世界的小白
当线程在系统内运行时，线程的调度具有一定的透明性，程序通常无法准确控制线程的轮换执行，但 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 线程通信的资料请关注脚本之家其它相关文章！