JAVA并发编程-线程间协作（Object监视器方法与Condition）

比眉伴天荒 2022-08-20 13:06 113阅读 0赞

JAVA并发编程第一篇

今天是2016年2月29日，四年一次，我琢磨着要写一篇blog...毕竟过了今天需要等四年啊！！！

说到线程间协作，不得不提到经典的生产者与消费者模型：有一个商品队列，生产者想队列中添加商品，消费者取出队列中的商品；显然，如果队列为空，消费者应该等待生产者产生商品才能消费；如果队列满了，生产者需要等待消费者消费之后才能生产商品。队列就是这个模型中的临界资源，当队列为空时，而消费者获得了该对象的锁，如果不释放，那么生产者无法获得对象锁，而消费者无法消费对象，就进入了死锁状态；反之队列满时，生产者不释放对象锁也会造成死锁。这是我们不希望看到的，所以就有了线程间协作来解决这个问题。

其实说到生产者与消费者模型，我们不能简单的知道怎么实现，而是需要知这种模型的使用场景：主要是为了复用和解耦，常见的消息框架（非常经典的一种生产者消费者模型的使用场景）ActiveMQ。发送端和接收端用Topic进行关联。

JAVA语言中，如何实现线程间协作呢？比较常见的方法就是利用Object.wait()，Object.notify()和Condition。

先看看这几个方法究竟有什么作用？为什么利用它们就可以实现线程间协作了呢？

首先分析一下wait()/notify()/notifyAll()这三个Object监视器方法，比较早的方法，JDK1.5之前：

1、上述三个方法都是Object类中的本地方法，且为final，无法被重写；且这三个方法都必须在同步块或者同步方法中才能执行；

2、当前线程必须拥有该对象的锁，才能执行wait()方法，wait()方法会阻塞当前线程，并且释放对象锁；

3、notify()方法可以唤醒一个（1/N）正在等待这个资源锁的线程，但是不保证被唤醒的线程一定可以获得这个对象锁。

4、notifyAll()方法可以唤醒所有正在等待这个资源锁的线程，然后让它们去竞争资源锁，具体哪个能拿到就不知道了。

下面看如何使用上述的wait()和notify()方法来实现生产者与消费者模式：

package org.zsl.learn.thread.join;
    
    import java.util.PriorityQueue;
    
    public class Producer implements Runnable{
    	private PriorityQueue<Integer> queue = null;
    	private int queueSize =0;
    	public Producer(PriorityQueue<Integer> queue,int queueSize){
    		this.queue=queue;
    		this.queueSize=queueSize;
    	}
    	
    	public void product(){
    		while(true){
    			synchronized (queue) {
    				System.out.println("当前队列中数据数量是："+queue.size());
    				while(queue.size()==queueSize){//对于生产者来说需要判断的是队列是否满了，如果满了就等待
    					System.out.println("队列已满，等待消费者消费....");
    					try {
    						queue.wait();
    					} catch (InterruptedException e) {
    						e.printStackTrace();
    						queue.notify(); //这里为什么加个notify呢？是为了防止死锁，线程出现问题时，也要释放对象锁。
    					}
    				}
    				//如果队列没满，那么就往队列中加入数据
    				queue.offer(1);
    				queue.notify();
    				try {
    					Thread.sleep(100);  //为什么加个休眠？是为了让我们可以在控制台看到生产者和消费者交替执行
    				} catch (InterruptedException e) {
    					// TODO Auto-generated catch block
    					e.printStackTrace();
    				}
    				System.out.println("向队列中插入一个数据，队列中剩余空间是："+(queueSize-queue.size()));
    			}
    		}
    	}
    
    	@Override
    	public void run() {
    		this.product();
    	}
    }

package org.zsl.learn.thread.join;
    
    import java.util.PriorityQueue;
    
    public class Consumer implements Runnable{
    	private PriorityQueue<Integer> queue = null;
    	
    	public Consumer(PriorityQueue<Integer> queue){
    		this.queue=queue;
    	}
    	
    	private void consume(){
    		while(true){
    			synchronized (queue) {	//首先锁定对象
    				//如果队列为空，那么消费者无法消费，必须等待生产者产生商品，所以需要释放对象锁，并让自己进入等待状态
    				System.out.println("当前队列中剩余数据个数："+queue.size());
    				while(queue.size()==0) {
    					System.out.println("队列为空，等待数据......");
    					try {
    						queue.wait();  //使用wait()这个方法的时候，对象必须是获取锁的状态，调用了这个方法后，线程会释放该对象锁
    					} catch (InterruptedException e) {
    						e.printStackTrace();
    						queue.notify();//这里为什么加个notify呢？是为了防止死锁，线程出现问题时，也要释放对象锁。
    					}
    				}
    				//如果不为空,取出第一个对象
    				queue.poll();
    				//注意notify()方法就是释放这个对象的锁，从而其他需要这个对象的线程中就会有一个能够获得锁，但是不能指定具体的线程
    				queue.notify();
    				try {
    					Thread.sleep(100);
    				} catch (InterruptedException e) {
    					// TODO Auto-generated catch block
    					e.printStackTrace();
    				}
    				System.out.println("消费一个数据后，队列中剩余数据个数："+queue.size());
    			}
    			
    		}
    	}
    
    
    	@Override
    	public void run() {
    		this.consume();
    	}
    	
    }

package org.zsl.learn.thread.join;
    
    import java.util.PriorityQueue;
    
    public class Test {
    	public static void main(String[] args) {
    		int queueSize = 20;
    		//这里可以回忆一下JVM中多线程共享内存的知识
    		PriorityQueue<Integer> queue = new PriorityQueue<>(queueSize);
    		
    		Consumer consumer = new Consumer(queue);
    		Producer producer = new Producer(queue, queueSize);
    		
    		new Thread(consumer).start();
    		new Thread(producer).start();
    	}
    }

\-----------------------------------------------------------------------------------------------------

下面我们再来分析一下JAVA中的Condition又是什么呢？java.util.concurrent.locks.Condition是为了替代Object监视器方法。那么我们不经就要问：Condition相比较Object监视器的三个方法有什么差别呢？下面我们就来对比看看，到底差别在哪里？下面这个表展示了它们方法之间的共性。

<table style="text-align:center"> 
 <tbody> 
  <tr> 
   <td style="text-align:center">&nbsp;</td> 
   <td style="text-align:center">Object</td> 
   <td style="text-align:center">Condititon</td> 
  </tr> 
  <tr> 
   <td style="text-align:center">休眠</td> 
   <td style="text-align:center">wait</td> 
   <td style="text-align:center">await</td> 
  </tr> 
  <tr> 
   <td style="text-align:center">唤醒一个线程</td> 
   <td style="text-align:center">notify</td> 
   <td style="text-align:center">signal</td> 
  </tr> 
  <tr> 
   <td style="text-align:center">唤醒所有线程</td> 
   <td style="text-align:center">notifyAll</td> 
   <td style="text-align:center">signalAll</td> 
  </tr> 
 </tbody> 
</table>

使用Condition中，使用Lock来替代Synchronized关键字来实现操作的原子性，实现对临界资源的加锁与解锁，同样的，Condition中提供的三个方法也需要在“同步块”中进行。Condition相比较而言，强大的地方在于它能够精确的控制多线程的休眠与唤醒（注意是唤醒，唤醒并不表示该线程一定能够得到资源锁），这个意思就是有A/B/C/D四个线程共享Z资源，如果A占用了Z，并且调用了b\_condition.notify()就可以释放资源唤醒B线程，而Object的nofity就无法保证B/C/D中会被唤醒哪一个了。其实多数线程间协作实用上述两种方式都可以实现，但是Sun推荐使用Condition来实现...我认为具体看你喜欢了，以及使用的熟练程度，除非你特别希望精确控制哪个线程被唤醒。

package org.zsl.learn.thread.join2;
    
    import java.util.PriorityQueue;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    public class Producer2 implements Runnable{
    	private PriorityQueue<Integer> queue = null;
    	private int queueSize =0;
    	private Lock lock = null;
    	private Condition consume=null;
    	private Condition produce=null;
    	
    	public Producer2(PriorityQueue<Integer> queue,int queueSize,Lock lock,Condition produce,Condition consume){
    		this.queue=queue;
    		this.queueSize=queueSize;
    		this.lock=lock;
    		this.consume=consume;
    		this.produce=produce;
    	}
    	
    	public void product(){
    		while(true){
    			lock.lock();
    			try{
    				while(queue.size()==queueSize){
    					System.out.println("队列满了，等待消费者消费...");
    					try {
    						produce.await();
    					} catch (InterruptedException e) {
    						// TODO Auto-generated catch block
    						e.printStackTrace();
    						consume.signal();
    					}
    				}
    				queue.offer(1);
    				System.out.println("向队列中插入了一个对象，队列的剩余空间是："+(queueSize-queue.size()));
    				consume.signal();
    				try {
    					Thread.sleep(100);
    				} catch (InterruptedException e) {
    					// TODO Auto-generated catch block
    					e.printStackTrace();
    				}
    				
    			}finally{
    				lock.unlock();
    			}
    		}
    	}
    
    	@Override
    	public void run() {
    		this.product();
    	}
    }

package org.zsl.learn.thread.join2;
    
    import java.util.PriorityQueue;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    
    public class Consumer2 implements Runnable{
    	private PriorityQueue<Integer> queue = null;
    	private Lock lock = null;
    	private Condition consume=null;
    	private Condition produce=null;
    			
    	
    	public Consumer2(PriorityQueue<Integer> queue,Lock lock,Condition produce,Condition consume){
    		this.queue=queue;
    		this.lock =lock;
    		this.consume = consume;
    		this.produce = produce;
    	}
    	
    	private void consume(){
    		while(true){
    			lock.lock();
    			try{
    				while(queue.size()==0){
    					System.out.println("队列为空，等待数据...");
    					try {
    						consume.await();
    					} catch (InterruptedException e) {
    						// TODO Auto-generated catch block
    						e.printStackTrace();
    						produce.signal();
    					}
    				}
    				queue.poll();
    				System.out.println("从队列中取出一个元素，队列剩余数量是："+queue.size());
    				produce.signal();
    				try {
    					Thread.sleep(100);
    				} catch (InterruptedException e) {
    					// TODO Auto-generated catch block
    					e.printStackTrace();
    				}
    			}finally{
    				lock.unlock();
    			}
    		}
    	}
    
    
    	@Override
    	public void run() {
    		this.consume();
    	}
    	
    }

其实在上述两个代码的实现结果中，如果不加上Thread.sleep()来让线程睡眠，我们看到的结果就像是单线程一样，生产者填满队列，消费者清空队列。为什么会这样呢？我们注意到，在“同步块”中，如果不是队列的临界值（0、maxSize）,仅仅是调用notify来唤醒一个等待该资源的线程，那么这个线程本身并没有进入等待状态，这个线程在释放这个锁之后会加入这个锁的竞争中，到底谁得到这个锁，其实也说不清楚，修改sleep的睡眠时间，可以看到从100毫秒到2000毫秒，设置不同的休眠时间，可以观察到生产者与消费者也不会出现交替进行，还是随机的。那么为什么要用Condition实现对确定线程的唤醒操作呢？唤醒了又不一定得到锁，这个需要使用到await()来让当前线程必须等到其他线程来唤醒才能控制生产者与消费者的交替执行。大家可以尝试一下：

在produce.signal()和consume.signal后面分别加上：consume.await()和produce.await即可实现生产者和消费者（多个线程也可以控制任意两个线程交替执行）的交替执行，这个呢，使用Object监视器方法在多个线程的情况下是不可能实现的，但是仅仅2个线程还是可以的。上述列子中，如果有多个消费者，那么如何在生产者完成生产后就只唤醒消费者线程呢？同样，用Condition实现就非常简单了，如果使用Object监视器类也可以实现，大家不妨想一下，但是相对复杂，编程过程中容易出现死锁。