线程间通信

墨蓝 2022-04-11 13:56 397阅读 0赞

一、引言

线程与线程之间不是相互独立的存在，它们彼此之间需要相互通信和协作。最典型的例子就是生产者-消费者问题。下面首先介绍 wait/notify 机制，并对实现该机制的两种方式：synchronized+wait-notify模式和 Lock+Condition 模式进行介绍，作为线程间通信与协作的基础。最后，对 Thread 类中的 join() 方法进行源码分析，并以宿主线程与寄生线程的协作为例进行说明。

二、使用 wait/notify/notifyAll 实现线程间通信的几点重要说明

1、Object是所有类的父类，它有5个方法组成了等待/通知机制的核心：notify()、notifyAll()、wait()、wait(long)和wait(long，int)。在Java中，所有的类都从Object继承而来，因此，所有的类都拥有这些共有方法可供使用。而且，由于他们都被声明为final，因此在子类中不能覆写任何一个方法。

2、在Java中，可以通过配合调用Object对象的 wait() 方法和 notify() 方法或 notifyAll() 方法来实现线程间的通信。上述三个方法均非Thread类中所声明的方法，而是Object类中声明的方法。原因是每个对象都拥有monitor（锁），所以让当前线程等待某个对象的锁，当然应该通过这个对象来操作，而不是用当前线程来操作，因为当前线程可能会等待多个线程的锁，如果通过线程来操作，就非常复杂了。

3、在线程中调用 wait() 方法，将阻塞等待其他线程的通知（其他线程调用 notify() 方法或 notifyAll() 方法），在线程中调用 notify() 方法或 notifyAll() 方法，将通知其他线程从 wait() 方法处返回。

4、wait()、notify()、notifyAll() 使用条件：
  1. 必须放在synchronized代码块中；
  2.     必须要锁住同一个对象，notify()方法才能唤醒等待的线程；
  3.     notify()只会随机的唤醒一个处于wait状态的线程；
  4.     notifyall()会全部唤醒所有处于wait线程，争夺到时间片的只有一个；
5、wait()方法

该方法用来将当前线程置入休眠状态，直到接到通知或被中断为止。在调用 wait() 之前，线程必须要获得该对象的对象级别锁，即只能在同步方法或同步块中调用 wait() 方法。 进入 wait() 方法后，当前线程释放锁。在从 wait() 返回前，线程与其他线程竞争重新获得锁。如果调用 wait() 时，没有持有适当的锁，则抛出IllegalMonitorStateException，它是RuntimeException的一个子类。

6、notify()

该方法也要在同步方法或同步块中调用，即在调用前，线程也必须要获得该对象的对象级别锁，如果调用 notify() 时没有持有适当的锁，也会抛出IllegalMonitorStateException。

该方法用来通知那些可能等待该对象的对象锁的其他线程。如果有多个线程等待，则线程规划器会随机挑选出其中一个 wait() 状态的线程来发出通知，并使它等待获取该对象的对象锁（notify后，当前线程不会马上释放该对象锁，wait所在的线程并不能马上获取该对象锁，要等到程序退出synchronized代码块后，当前线程才会释放锁，wait所在的线程也才可以获取该对象锁），但不惊动其他同样在等待被该对象notify的线程们。当第一个获得了该对象锁的 wait 线程运行完毕以后，它会释放掉该对象锁，此时如果该对象没有再次使用notify语句，则即便该对象已经空闲，其他wait状态等待的线程由于没有得到该对象的通知，会继续阻塞在wait状态，直到这个对象发出一个notify或notifyAll。这里需要注意：它们等待的是被notify或notifyAll，而不是锁。这与下面的notifyAll()方法执行后的情况不同。

7、notifyAll()

该方法与 notify() 方法的工作方式相同，重要的一点差异是：

notifyAll() 使所有原来在该对象上 wait 的线程统统退出 wait 的状态（即全部被唤醒，不再等待notify或notifyAll，但由于此时还没有获取到该对象锁，因此还不能继续往下执行），变成等待获取该对象上的锁， 一旦该对象锁被释放（notifyAll线程退出调用了notifyAll的synchronized代码块的时候），它们就会去竞争。如果其中一个线程获得了该对象锁，它就会继续往下执行，在它退出synchronized代码块，释放锁后，其他的已经被唤醒的线程将会继续竞争获取该锁，一直进行下去，直到所有被唤醒的线程都执行完毕。

8、wait(long) 和 wait(long，int)

这两个方法是设置等待超时时间的，后者在超值时间上加上ns，精度也难以达到，因此，该方法很少使用。对于前者，如果在等待的线程接到通知或被中断之前，已经超过了指定的毫秒数，则它通过竞争重新获得锁，并从 wait(long) 返回。另外，需要知道，如果设置了超时时间，当 wait() 返回时，我们不能确定它是因为接到了通知还是因为超时而返回的，因为 wait() 方法不会返回任何相关的信息。但一般可以通过设置标志位来判断，在notify之前改变标志位的值，在 wait() 方法后读取该标志位的值来判断，当然为了保证notify不被遗漏，我们还需要另外一个标志位来循环判断是否调用 wait() 方法。

9、小结
  1. a.    wait()、notify() 和 notifyAll()方法是 本地方法，并且为 final 方法，无法被重写；
  2. b.    调用某个对象的 wait() 方法能让 当前线程阻塞，并且当前线程必须拥有此对象的monitor（即锁）；
  3. c.    调用某个对象的 notify() 方法能够唤醒 一个正在等待这个对象的monitor的线程，如果有多个线程都
  4.     在等待这个对象的monitor，则只能唤醒其中一个线程；
  5. d.    调用notifyAll()方法能够唤醒所有正在等待这个对象的monitor的线程。

三、生产者-消费者模型

线程与线程之间相互通信和协作，最经典的例子就是 生产者-消费者模型：
　　
在下面的例子中，虽然两个线程实现了通信，但是凭借 线程B 不断地通过 while语句轮询 来检测某一个条件，这样会导致CPU的浪费。因此，需要一种机制来减少 CPU资源 的浪费，而且还能实现多个线程之间的通信，即 wait/notify 机制 。

  1. //资源类
  2. class MyList { 
  4.     //临界资源
  5.     private volatile List<String> list = new ArrayList<String>();
  7.     public void add() { 
  8.         list.add("abc");
  9.     }
  11.     public int size() { 
  12.         return list.size();
  13.     }
  14. }
  16. // 线程A
  17. class ThreadA extends Thread { 
  19.     private MyList list;//临界资源
  21.     public ThreadA(MyList list,String name) { 
  22.         super(name);
  23.         this.list = list;
  24.     }
  26.     @Override
  27.     public void run() { 
  28.         try { 
  29.             for (int i = 0; i < 3; i++) { 
  30.                 list.add();
  31.                 System.out.println("添加了" + (i + 1) + "个元素");
  32.                 Thread.sleep(1000);
  33.             }
  34.         } catch (InterruptedException e) { 
  35.             e.printStackTrace();
  36.         }
  37.     }
  38. }
  40. //线程B
  41. class ThreadB extends Thread { 
  43.     private MyList list;//临界资源
  45.     public ThreadB(MyList list,String name) { 
  46.         super(name);
  47.         this.list = list;
  48.     }
  50.     @Override
  51.     public void run() { 
  52.         try { 
  53.             while (true) {           // while 语句轮询
  54.                 if (list.size() == 2) { 
  55.                     System.out.println("==2了，线程b要退出了！");
  56.                     throw new InterruptedException();
  57.                 }
  58.             }
  59.         } catch (InterruptedException e) { 
  60.             e.printStackTrace();
  61.         }
  62.     }
  63. }
  65. //测试
  66. public class Test { 
  68.     public static void main(String[] args) { 
  70.         MyList service = new MyList();
  72.         ThreadA a = new ThreadA(service,"A");
  73.         ThreadB b = new ThreadB(service,"B");
  75.         a.start();
  76.         b.start();
  77.     }
  78. }
  79. /* Output(输出结果不唯一): 添加了1个元素 添加了2个元素 ==2了，线程b要退出了！ java.lang.InterruptedException at test.ThreadB.run(Test.java:57) 添加了3个元素 *///:~

四、 wait/notify 机制

1、方法调用与线程状态关系

每个锁对象都有两个队列，一个是就绪队列，一个是阻塞队列。就绪队列存储了已就绪（将要竞争锁）的线程，阻塞队列存储了被阻塞的线程。当一个阻塞线程被唤醒后，才会进入就绪队列，进而等待CPU的调度；反之，当一个线程被 wait 后，就会进入阻塞队列，等待被唤醒。
在这里插入图片描述

2、使用举例

  1. public class Test { 
  3.     public static Object object = new Object();
  5.     public static void main(String[] args) throws InterruptedException { 
  6.         Thread1 thread1 = new Thread1();
  7.         Thread2 thread2 = new Thread2();
  9.         thread1.start();
  11.         Thread.sleep(2000);//main线程休眠2秒
  13.         thread2.start();
  14.     }
  16.     static class Thread1 extends Thread { 
  17.         @Override
  18.         public void run() { 
  19.             synchronized (object) { 
  20.                 System.out.println("线程" + Thread.currentThread().getName()
  21.                         + "获取到了锁...");
  22.                 try { 
  23.                     System.out.println("线程" + Thread.currentThread().getName()
  24.                             + "阻塞并释放锁...");
  25.                     object.wait();
  26.                 } catch (InterruptedException e) { 
  27.                 }
  28.                 System.out.println("线程" + Thread.currentThread().getName()
  29.                         + "执行完成...");
  30.             }
  31.         }
  32.     }
  34.     static class Thread2 extends Thread { 
  35.         @Override
  36.         public void run() { 
  37.             synchronized (object) { 
  38.                 System.out.println("线程" + Thread.currentThread().getName()
  39.                         + "获取到了锁...");
  40.                 object.notify();
  41.                 System.out.println("线程" + Thread.currentThread().getName()
  42.                         + "唤醒了正在wait的线程...");
  43.             }
  44.             System.out
  45.                     .println("线程" + Thread.currentThread().getName() + "执行完成...");
  46.         }
  47.     }
  48. }
  49. /* Output: 线程Thread-0获取到了锁... 线程Thread-0阻塞并释放锁... 线程Thread-1获取到了锁... 线程Thread-1唤醒了正在wait的线程... 线程Thread-1执行完成... 线程Thread-0执行完成... *///:~

3、多个同类型线程的场景（wait 的条件发生变化）

  1. //资源类
  2. class ValueObject { 
  3.     public static List<String> list = new ArrayList<String>();
  4. }
  6. //元素添加线程
  7. class ThreadAdd extends Thread { 
  9.     private String lock;
  11.     public ThreadAdd(String lock,String name) { 
  12.         super(name);
  13.         this.lock = lock;
  14.     }
  16.     @Override
  17.     public void run() { 
  18.         synchronized (lock) { 
  19.             ValueObject.list.add("anyString");
  20.             lock.notifyAll();               // 唤醒所有 wait 线程
  21.         }
  22.     }
  23. }
  25. //元素删除线程
  26. class ThreadSubtract extends Thread { 
  28.     private String lock;
  30.     public ThreadSubtract(String lock,String name) { 
  31.         super(name);
  32.         this.lock = lock;
  33.     }
  35.     @Override
  36.     public void run() { 
  37.         try { 
  38.             synchronized (lock) { 
  39.                 if (ValueObject.list.size() == 0) { 
  40.                     System.out.println("wait begin ThreadName=" + Thread.currentThread().getName());
  41.                     lock.wait();
  42.                     System.out.println("wait end ThreadName=" + Thread.currentThread().getName());
  43.                 }
  44.                 ValueObject.list.remove(0);
  45.                 System.out.println("list size=" + ValueObject.list.size());
  46.             }
  47.         } catch (InterruptedException e) { 
  48.             e.printStackTrace();
  49.         }
  50.     }
  51. }
  53. //测试类
  54. public class Run { 
  55.     public static void main(String[] args) throws InterruptedException { 
  57.         //锁对象
  58.         String lock = new String("");
  60.         ThreadSubtract subtract1Thread = new ThreadSubtract(lock,"subtract1Thread");
  61.         subtract1Thread.start();
  63.         ThreadSubtract subtract2Thread = new ThreadSubtract(lock,"subtract2Thread");
  64.         subtract2Thread.start();
  66.         Thread.sleep(1000);
  68.         ThreadAdd addThread = new ThreadAdd(lock,"addThread");
  69.         addThread.start();
  71.     }
  72. }/* Output: wait begin ThreadName=subtract1Thread wait begin ThreadName=subtract2Thread wait end ThreadName=subtract2Thread list size=0 wait end ThreadName=subtract1Thread Exception in thread "subtract1Thread" java.lang.IndexOutOfBoundsException: Index: 0, Size: 0 at java.util.ArrayList.rangeCheck(Unknown Source) at java.util.ArrayList.remove(Unknown Source) at test.ThreadSubtract.run(Run.java:49) *///:~

当 线程subtract1Thread 被唤醒后，将从 wait 处继续执行。但由于 线程subtract2Thread 先获取到锁得到运行，导致 线程subtract1Thread 的 wait 的条件发生变化（不再满足），而 线程subtract1Thread 却毫无所知，导致异常产生。
像这种有多个相同类型的线程场景，为防止 wait 的条件发生变化而导致的线程异常终止，我们在阻塞线程被唤醒的同时还必须对 wait 的条件进行额外的检查，如下所示：

  1. //元素删除线程
  2. class ThreadSubtract extends Thread { 
  4.     private String lock;
  6.     public ThreadSubtract(String lock,String name) { 
  7.         super(name);
  8.         this.lock = lock;
  9.     }
  11.     @Override
  12.     public void run() { 
  13.         try { 
  14.             synchronized (lock) { 
  15.                 while (ValueObject.list.size() == 0) {     //将 if 改成 while 
  16.                     System.out.println("wait begin ThreadName=" + Thread.currentThread().getName());
  17.                     lock.wait();
  18.                     System.out.println("wait end ThreadName=" + Thread.currentThread().getName());
  19.                 }
  20.                 ValueObject.list.remove(0);
  21.                 System.out.println("list size=" + ValueObject.list.size());
  22.             }
  23.         } catch (InterruptedException e) { 
  24.             e.printStackTrace();
  25.         }
  26.     }
  27. }/* Output: wait begin ThreadName=subtract1Thread wait begin ThreadName=subtract2Thread wait end ThreadName=subtract2Thread list size=0 wait end ThreadName=subtract1Thread wait begin ThreadName=subtract1Thread *///:~

只需将 线程类ThreadSubtract 的 run()方法中的 if 条件 改为 while 条件 即可。

四、 Condition

Condition 是在java 1.5中出现的，它用来替代传统的 Object 的wait()/notify()实现线程间的协作，它的使用依赖于 Lock。Condition、Lock 和 Thread 三者之间的关系如下图所示。相比使用Object的wait()/notify()，使用Condition的await()/signal()这种方式能够更加安全和高效地实现线程间协作。Condition是个接口，基本的方法就是await()和signal()方法。Condition依赖于Lock接口，生成一个Condition的基本代码是lock.newCondition() 。 必须要注意的是，Condition 的 await()/signal() 使用都必须在lock保护之内，也就是说，必须在lock.lock()和lock.unlock之间才可以使用。事实上，Conditon的await()/signal() 与 Object的wait()/notify() 有着天然的对应关系：

Conditon中的await()对应Object的wait()；
Condition中的signal()对应Object的notify()；
Condition中的signalAll()对应Object的notifyAll()。
在这里插入图片描述

使用Condition往往比使用传统的通知等待机制(Object的wait()/notify())要更灵活、高效，例如，我们可以使用多个Condition实现通知部分线程：

  1. // 线程 A
  2. class ThreadA extends Thread { 
  3.     private MyService service;
  4.     public ThreadA(MyService service) { 
  5.         super();
  6.         this.service = service;
  7.     }
  8.     @Override
  9.     public void run() { 
  10.         service.awaitA();
  11.     }
  12. }
  13. // 线程 B
  14. class ThreadB extends Thread { 
  15.     private MyService service;
  16.     public ThreadB(MyService service) { 
  17.         super();
  18.         this.service = service;
  19.     }
  20.     @Override
  21.     public void run() { 
  22.         service.awaitB();
  23.     }
  24. }
  26. class MyService { 
  27.     private Lock lock = new ReentrantLock();
  28.     // 使用多个Condition实现通知部分线程
  29.     public Condition conditionA = lock.newCondition();
  30.     public Condition conditionB = lock.newCondition();
  32.     public void awaitA() { 
  33.         lock.lock();
  34.         try { 
  35.             System.out.println("begin awaitA时间为" + System.currentTimeMillis()
  36.                     + " ThreadName=" + Thread.currentThread().getName());
  37.             conditionA.await();
  38.             System.out.println(" end awaitA时间为" + System.currentTimeMillis()
  39.                     + " ThreadName=" + Thread.currentThread().getName());
  40.         } catch (InterruptedException e) { 
  41.             e.printStackTrace();
  42.         } finally { 
  43.             lock.unlock();
  44.         }
  45.     }
  47.     public void awaitB() { 
  48.         lock.lock();
  49.         try { 
  50.             System.out.println("begin awaitB时间为" + System.currentTimeMillis()
  51.                     + " ThreadName=" + Thread.currentThread().getName());
  52.             conditionB.await();
  53.             System.out.println(" end awaitB时间为" + System.currentTimeMillis()
  54.                     + " ThreadName=" + Thread.currentThread().getName());
  55.         } catch (InterruptedException e) { 
  56.             e.printStackTrace();
  57.         } finally { 
  58.             lock.unlock();
  59.         }
  60.     }
  62.     public void signalAll_A() { 
  63.         try { 
  64.             lock.lock();
  65.             System.out.println(" signalAll_A时间为" + System.currentTimeMillis()
  66.                     + " ThreadName=" + Thread.currentThread().getName());
  67.             conditionA.signalAll();
  68.         } finally { 
  69.             lock.unlock();
  70.         }
  71.     }
  73.     public void signalAll_B() { 
  74.         try { 
  75.             lock.lock();
  76.             System.out.println(" signalAll_B时间为" + System.currentTimeMillis()
  77.                     + " ThreadName=" + Thread.currentThread().getName());
  78.             conditionB.signalAll();
  79.         } finally { 
  80.             lock.unlock();
  81.         }
  82.     }
  83. }
  85. // 测试
  86. public class Run { 
  87.     public static void main(String[] args) throws InterruptedException { 
  88.         MyService service = new MyService();
  90.         ThreadA a = new ThreadA(service);
  91.         a.setName("A");
  92.         a.start();
  94.         ThreadB b = new ThreadB(service);
  95.         b.setName("B");
  96.         b.start();
  98.         Thread.sleep(3000);
  99.         service.signalAll_A();
  100.     }
  101. }

输出结果如下图所示，我们可以看到只有线程A被唤醒，线程B仍然阻塞。
在这里插入图片描述
实际上，Condition 实现了一种分组机制，将所有对临界资源进行访问的线程进行分组，以便实现线程间更精细化的协作，例如通知部分线程。我们可以从上面例子的输出结果看出，只有conditionA范围内的线程A被唤醒，而conditionB范围内的线程B仍然阻塞。

六、 生产者-消费者模型

等待/通知机制 最经典的应用就是 生产者-消费者模型。下面以多生产者-多消费者问题为背景，分别运用两种模式 synchronized+wait-notify 模式和 Lock+Condition 模式实现 wait-notify 机制。

传统实现方式
  1. //资源类
  2. class MyStack { 
  3.     // 共享队列
  4.     private List list = new ArrayList();
  6.     // 生产
  7.     public synchronized void push() { 
  8.         try { 
  9.             while (list.size() == 1) {     // 多个生产者
  10.                 System.out.println("队列已满，线程 "
  11.                         + Thread.currentThread().getName() + " 呈wait状态...");
  12.                 this.wait();
  13.             }
  14.             list.add("anyString=" + Math.random());
  15.             System.out.println("线程 " + Thread.currentThread().getName()
  16.                     + " 生产了，队列已满...");
  17.             this.notifyAll();                   // 防止生产者仅通知生产者
  18.         } catch (InterruptedException e) { 
  19.             e.printStackTrace();
  20.         }
  21.     }
  23.     // 消费
  24.     public synchronized String pop() { 
  25.         String returnValue = "";
  26.         try { 
  27.             while (list.size() == 0) {               // 多个消费者
  28.                 System.out.println("队列已空，线程 "
  29.                         + Thread.currentThread().getName() + " 呈wait状态...");
  30.                 this.wait();
  31.             }
  32.             returnValue = "" + list.get(0);
  33.             list.remove(0);
  34.             System.out.println("线程 " + Thread.currentThread().getName()
  35.                     + " 消费了，队列已空...");
  36.             this.notifyAll();                   // 防止消费者仅通知消费者
  37.         } catch (InterruptedException e) { 
  38.             e.printStackTrace();
  39.         }
  40.         return returnValue;
  41.     }
  42. }
  44. //生产者
  45. class P_Thread extends Thread { 
  47.     private MyStack myStack;
  49.     public P_Thread(MyStack myStack,String name) { 
  50.         super(name);
  51.         this.myStack = myStack;
  52.     }
  54.     public void pushService() { 
  55.         myStack.push();
  56.     }
  58.     @Override
  59.     public void run() { 
  60.         while (true) {      
  61.             myStack.push();
  62.         }
  63.     }
  64. }
  66. //消费者
  67. class C_Thread extends Thread { 
  69.     private MyStack myStack;
  71.     public C_Thread(MyStack myStack,String name) { 
  72.         super(name);
  73.         this.myStack = myStack;
  74.     }
  76.     @Override
  77.     public void run() { 
  78.         while (true) { 
  79.             myStack.pop();
  80.         }
  81.     }
  82. }
  84. //测试类
  85. public class Run { 
  86.     public static void main(String[] args) throws InterruptedException { 
  87.         MyStack myStack = new MyStack();
  89.         P_Thread pThread1 = new P_Thread(myStack, "P1");
  90.         P_Thread pThread2 = new P_Thread(myStack, "P2");
  91.         P_Thread pThread3 = new P_Thread(myStack, "P3");
  92.         P_Thread pThread4 = new P_Thread(myStack, "P4");
  93.         P_Thread pThread5 = new P_Thread(myStack, "P5");
  94.         P_Thread pThread6 = new P_Thread(myStack, "P6");
  95.         pThread1.start();
  96.         pThread2.start();
  97.         pThread3.start();
  98.         pThread4.start();
  99.         pThread5.start();
  100.         pThread6.start();
  102.         C_Thread cThread1 = new C_Thread(myStack, "C1");
  103.         C_Thread cThread2 = new C_Thread(myStack, "C2");
  104.         C_Thread cThread3 = new C_Thread(myStack, "C3");
  105.         C_Thread cThread4 = new C_Thread(myStack, "C4");
  106.         C_Thread cThread5 = new C_Thread(myStack, "C5");
  107.         C_Thread cThread6 = new C_Thread(myStack, "C6");
  108.         C_Thread cThread7 = new C_Thread(myStack, "C7");
  109.         C_Thread cThread8 = new C_Thread(myStack, "C8");
  110.         cThread1.start();
  111.         cThread2.start();
  112.         cThread3.start();
  113.         cThread4.start();
  114.         cThread5.start();
  115.         cThread6.start();
  116.         cThread7.start();
  117.         cThread8.start();
  118.     }
  119. }/* Output: 线程 P1 生产了，队列已满... 队列已满，线程 P1 呈wait状态... 线程 C5 消费了，队列已空... 队列已空，线程 C5 呈wait状态... 队列已空，线程 C8 呈wait状态... 队列已空，线程 C2 呈wait状态... 队列已空，线程 C7 呈wait状态... 队列已空，线程 C4 呈wait状态... 队列已空，线程 C6 呈wait状态... 队列已空，线程 C3 呈wait状态... 队列已空，线程 C1 呈wait状态... 线程 P6 生产了，队列已满... 队列已满，线程 P6 呈wait状态... 队列已满，线程 P5 呈wait状态... 队列已满，线程 P4 呈wait状态... ... *///:~

对于生产者-消费者问题，有两个要点需要注意：

在多个同类型线程（多个生产者线程或者消费者线程）的场景中，为防止wait的条件发生变化而导致线程异常终止，我们在阻塞线程被唤醒的同时还必须对wait的条件进行额外的检查，即 使用 while 循环代替 if 条件；

在多个同类型线程（多个生产者线程或者消费者线程）的场景中，为防止生产者(消费者)唤醒生产者(消费者)，保证生产者和消费者互相唤醒，需要 使用 notify 替代 notifyAll.

使用 Condition 实现方式
  1. // 线程A
  2. class MyThreadA extends Thread { 
  4.     private MyService myService;
  6.     public MyThreadA(MyService myService, String name) { 
  7.         super(name);
  8.         this.myService = myService;
  9.     }
  11.     @Override
  12.     public void run() { 
  13.         while (true)
  14.             myService.set();
  15.     }
  16. }
  18. // 线程B
  19. class MyThreadB extends Thread { 
  21.     private MyService myService;
  23.     public MyThreadB(MyService myService, String name) { 
  24.         super(name);
  25.         this.myService = myService;
  26.     }
  28.     @Override
  29.     public void run() { 
  30.         while (true)
  31.             myService.get();
  32.     }
  33. }
  35. // 资源类
  36. class MyService { 
  38.     private ReentrantLock lock = new ReentrantLock();
  39.     private Condition conditionA = lock.newCondition();   // 生产线程
  40.     private Condition conditionB = lock.newCondition();  // 消费线程
  41.     private boolean hasValue = false;
  43.     public void set() { 
  44.         try { 
  45.             lock.lock();
  46.             while (hasValue == true) { 
  47.                 System.out.println("[生产线程] " + " 线程"
  48.                         + Thread.currentThread().getName() + " await...");
  49.                 conditionA.await();
  50.             }
  51.             System.out.println("[生产中] " + " 线程" + Thread.currentThread().getName() + " 生产★");
  52.             Thread.sleep(1000);
  53.             hasValue = true;
  54.             System.out.println("线程" + Thread.currentThread().getName() + " 生产完毕...");
  55.             System.out.println("[唤醒所有消费线程] " + " 线程"
  56.                     + Thread.currentThread().getName() + "...");
  57.             conditionB.signalAll();
  58.         } catch (InterruptedException e) { 
  59.             e.printStackTrace();
  60.         } finally { 
  61.             lock.unlock();
  62.         }
  63.     }
  65.     public void get() { 
  66.         try { 
  67.             lock.lock();
  68.             while (hasValue == false) { 
  69.                 System.out.println("[消费线程] " + " 线程"
  70.                         + Thread.currentThread().getName() + " await...");
  71.                 conditionB.await();
  72.             }
  73.             System.out.println("[消费中] " + " 线程"
  74.                     + Thread.currentThread().getName() + " 消费☆");
  75.             Thread.sleep(1000);
  76.             System.out.println("线程" + Thread.currentThread().getName() + " 消费完毕...");
  77.             hasValue = false;
  78.             System.out.println("[唤醒所有生产线程] " + " 线程"
  79.                     + Thread.currentThread().getName() + "...");
  80.             conditionA.signalAll();
  81.         } catch (InterruptedException e) { 
  82.             e.printStackTrace();
  83.         } finally { 
  84.             lock.unlock();
  85.         }
  86.     }
  87. }
  89. public class Run { 
  90.     public static void main(String[] args) throws InterruptedException { 
  91.         MyService service = new MyService();
  93.         MyThreadA[] threadA = new MyThreadA[10];
  94.         MyThreadB[] threadB = new MyThreadB[10];
  96.         for (int i = 0; i < 10; i++) { 
  97.             threadA[i] = new MyThreadA(service, "ThreadA-" + i);
  98.             threadB[i] = new MyThreadB(service, "ThreadB-" + i);
  99.             threadA[i].start();
  100.             threadB[i].start();
  101.         }
  102.     }
  103. }/* Output: [生产中] 线程ThreadA-0 生产★ 线程ThreadA-0 生产完毕... [唤醒所有消费线程] 线程ThreadA-0... [生产线程] 线程ThreadA-0 await... [消费中] 线程ThreadB-0 消费☆ 线程ThreadB-0 消费完毕... [唤醒所有生产线程] 线程ThreadB-0... [消费线程] 线程ThreadB-0 await... [生产中] 线程ThreadA-1 生产★ 线程ThreadA-1 生产完毕... [唤醒所有消费线程] 线程ThreadA-1... [生产线程] 线程ThreadA-1 await... [消费中] 线程ThreadB-1 消费☆ 线程ThreadB-1 消费完毕... [唤醒所有生产线程] 线程ThreadB-1... [消费线程] 线程ThreadB-1 await... [生产中] 线程ThreadA-2 生产★ 线程ThreadA-2 生产完毕... [唤醒所有消费线程] 线程ThreadA-2... ... *///:~

发表评论

表情：
评论列表 （有 0 条评论，397人围观）

还没有评论，来说两句吧...

相关阅读

    相关 线程间通信ThreadLocal

    ThreadLocal，即线程变量，是一个以ThreadLocal对象为键、任意对象为值的存储结构。这 个结构被附带在线程上，也就是说一个线程可以根据一个ThreadLocal

    青旅半醒青旅半醒/ 0 赞/ 301 阅读

    相关 线程间通信

    为什么需要线程通讯 线程是操作系统调度的最小单位，有自己的栈空间，可以按照既定的代码逐步的执行，但是如果每个线程间都孤立的运行，那就会造资源浪费。所以在现实中，我们需要这

    心已赠人心已赠人/ 0 赞/ 360 阅读

    相关 Java线程间通信

    上述例题无条件的阻塞了其他线程异步访问某个方法。Java对象中隐式管程的应用是很强大的，但是你可以通过进程间通信达到更微妙的境界。这在Java中是尤为简单的。 像前面

    古城微笑少年丶古城微笑少年丶/ 0 赞/ 387 阅读

    相关 线程间通信

    线程和线程之间不是独立的个体，它们彼此之间可以互相通信和协作。 线程通信就是在线程之间传递信息，保证他们能够协同工作。在线程间进行通信后，系统之间的交互性会更强大，在大大提高

    Myth丶恋晨Myth丶恋晨/ 0 赞/ 368 阅读

    相关 线程间通信

    一、引言 线程与线程之间不是相互独立的存在，它们彼此之间需要相互通信和协作。最典型的例子就是生产者-消费者问题。下面首先介绍 wait/notify 机制，并对实现该机制

    墨蓝墨蓝/ 0 赞/ 398 阅读

    相关 线程间通信

    注意： 必须在同步方法中使用wait和notify方法，因为执行wait和notify的前提条件是必须持有同步方法（或块）的monitor的所有权，否则将会抛出异常

    女爷i女爷i/ 0 赞/ 493 阅读