多线程编程-线程间通信.wait/notify（四）

蔚落 2022-06-01 01:12 296阅读 0赞

多个线程通过共同访问一个变量的方式，也可以实现线程间通信，通过这种方式，线程要主动去读取一个共享变量，一方面需要加同步，另一方面花费了读取变量的时间，但是读取到的值是不是要想要的，并不能确定。

而等待/通知机制，可以更好解决线程间数据读取时机的问题。

1，不使用等待/通知机制的线程间通信

测试代码：

public class MyList {
    	private List list = new ArrayList();
    	public void add(){
    		list.add("thread trans data");
    	}
    	
    	public int size(){
    		return list.size();
    	}
    }

public class ThreadA extends Thread{
    	private MyList list;
    	
    	public ThreadA(MyList list){
    		super();
    		this.list = list;
    	}
    	
    	@Override
    	public void run(){
    		try{
    			for(int i =0; i<10; i++){
    				list.add();
    				System.out.println("add item "+(i+1)+"...");
    				Thread.sleep(1000);
    			}
    		}catch(InterruptedException e){
    			e.printStackTrace();
    		}
    	}
    }

public class ThreadB extends Thread{
    	private MyList list;
    	
    	public ThreadB(MyList list){
    		super();
    		this.list = list;
    	}
    	
    	@Override
    	public void run(){
    		try{
    			while(true){
    				if(list.size() == 5){
    					System.out.println("size == 5,thread exit");
    					throw new InterruptedException();
    				}
    			}
    		}catch(InterruptedException e){
    			e.printStackTrace();
    		}
    	}
    }

public class RunDemo {
    
    	public static void main(String[] args) {
    		MyList run = new MyList();
    		ThreadA tA= new ThreadA(run);
    		tA.setName("A");
    		tA.start();
    		ThreadB  tB =new ThreadB(run);
    		tB.setName("B");
    		tB.start();
    	}
    }

运行结果：

add item 1...  
add item 2...  
add item 3...  
add item 4...  
add item 5...  
size == 5,thread exit  
java.lang.InterruptedException  
at ThreadB.run(ThreadB.java:16)  
add item 6...  
add item 7...  
add item 8...  
add item 9...  
add item 10...  
  
虽然两个线程间实现了通信，但是一个弊端是线程B要通过while循环来检测某一个条件，这样会浪费cpu时间，而且如果轮询的间隔时间过大，会取不到想要的结果。wait/notify机制可以解决这个弊端。

2，等待/通知机制的实现

方法wait()的作用是使当前执行代码的线程进入等待，它是Object类的方法，该方法用来将当前线程置入“预执行队列”中，并在wait()所在的代码行停止执行，直到接到通知或被中断为止。

在调用wait()之前，线程需要获得该对象的对象级别锁，所以只能在同步方法或同步块中调用wait()方法。执行wait()后，当前线程释放锁。在从wait()返回前，线程与其他线程竞争重新获得锁。如果调用wait()时没有持有适当的锁，抛出IllegalMonitorstateException。

方法notify()也要在同步方法或同步块中调用，在调用前，线程也要获得该对象的对象级别锁。该方法用来通知那些可能等待该对象的对象锁的其他线程，如果有多个线程等待，则有线程规划器随机挑选出一个呈wait状态的线程，对其发出notify，并使他获取该对象的对象锁。

在执行notify()方法后，当前线程不会马上释放该对象锁，呈wait状态的线程也不能马上获取该对象锁，要等到执行notify()方法的线程将程序执行完，也即是退出synchronized代码块后，当前线程才会释放锁，而呈wait状态的线程才可能获取该对象锁。

当第一个获取该对象锁的wait线程运行完毕后，他会释放该对象锁，此时如果该对象没有再次使用notify语句，即使该对象已经空闲，其他wait状态的线程由于没有得到该对象的通知，还会继续阻塞在wait状态，直到这个对象发出一个notify/notifyall。

测试代码（修改 1）中的ThreadA，ThreadB的代码）：

public class ThreadA extends Thread{
    	private MyList list;
    	
    	public ThreadA(MyList list){
    		super();
    		this.list = list;
    	}
    	
    	@Override
    	public void run(){
    		try{
    			synchronized(list){
    				for(int i =0; i<10; i++){
    					list.add();
    					if(list.size() == 5){
    						System.out.println("已发出通知。");
    						list.notify();
    					}
    					System.out.println("add item "+(i+1)+"...");
    					Thread.sleep(1000);
    				}			
    			}
    
    		}catch(InterruptedException e){
    			e.printStackTrace();
    		}
    	}
    }

public class ThreadB extends Thread{
    	private MyList list;
    	
    	public ThreadB(MyList list){
    		super();
    		this.list = list;
    	}
    	
    	@Override
    	public void run(){
    		try{
    			synchronized(list){
    				if(list.size() != 5){
    					System.out.println("wait begin="+System.currentTimeMillis());
    					list.wait();
    					System.out.println("wait end="+System.currentTimeMillis());
    				}				
    			}
    		}catch(InterruptedException e){
    			e.printStackTrace();
    		}
    	}
    }

public class RunDemo {
    
    	public static void main(String[] args) {
    		MyList run = new MyList();
    		try {
    			ThreadB  tB =new ThreadB(run);
    			tB.setName("B");
    			tB.start();				
    			Thread.sleep(500);
    			ThreadA tA= new ThreadA(run);
    			tA.setName("A");
    			tA.start();				
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    	}
    }

运行结果：

wait begin=1516089526512  
add item 1...  
add item 2...  
add item 3...  
add item 4...  
已发出通知。  
add item 5...  
add item 6...  
add item 7...  
add item 8...  
add item 9...  
add item 10...  
wait end=1516089537079

日志信息waitend在最后输出，也说明notify()方法执行后并不立即释放锁。

这两个方法须用在被synchronized同步的object的临界区内，调用wait()方法是处于临界区内的线程进入等待，同时释放被同步对象的锁。而notify唤醒一个因调用wait而处于阻塞状态的线程，使其进入就绪状态。被重新唤醒的线程会试图重新获得临界区的控制权，即锁，并继续执行临界区内wait之后的代码。

Notifyall()方法可以使所有正在等待队列中等待同一共享资源的“全部”线程从等待状态退出，进入可运行状态。此时，优先级高的线程可能先执行，也可能随机执行，具体取决于虚拟机的实现。

在执行同步代码块的过程中，遇到异常而导致线程终止，锁也会被释放。

3， 生产者 / 消费者模式的实现

等待/通知机制最经典的案例是生产者/消费者模式。

3.1 一个生产者与一个消费者：操作值  
测试代码：

public class ValueObject {
    	public static String value = "";
    }

public class Producer {
    	private Object lock;
    	public Producer(Object lock){
    		this.lock = lock;
    	}
    	public void setValue(){
    		try {
    			synchronized(lock){
    				if(!ValueObject.value.equals("")){
    					lock.wait();
    				}
    				String value = System.currentTimeMillis()+"_"+System.nanoTime();
    				System.out.println("set value is "+value);
    				ValueObject.value = value;
    				lock.notify();
    			}
    		} catch (InterruptedException e) {
    		}
    	}
    }

public class Consumer {
    	private Object lock;
    	public Consumer(Object lock){
    		this.lock = lock;
    	}
    	public void getValue(){
    		try {
    			synchronized(lock){
    				if(ValueObject.value.equals("")){
    					lock.wait();
    				}
    				System.out.println("get value is "+ValueObject.value);
    				ValueObject.value = "";
    				lock.notify();
    			}
    		} catch (InterruptedException e) {
    		}
    	}
    }

public class ThreadP extends Thread{
    	private Producer p;
    	public ThreadP(Producer p){
    		this.p = p;
    	}
    	@Override
    	public void run() {
    		while(true){
    			p.setValue();
    		}
    	}
    }

public class ThreadC extends Thread{
    	private Consumer c;
    	public ThreadC(Consumer c){
    		this.c = c;
    	}
    	@Override
    	public void run() {
    		while(true){
    			c.getValue();
    		}
    	}
    }

public class RunDemo {
    
    	public static void main(String[] args) {
    		Object obj = new Object();
    		Producer p = new Producer(obj);
    		Consumer c = new Consumer(obj);
    		ThreadP tp = new ThreadP(p);
    		ThreadC tc = new ThreadC(c);
    		tp.start();
    		tc.start();
    	}
    }

运行结果：

...

set value is 1516092614318\_27978884503426  
get value is 1516092614318\_27978884503426  
set value is 1516092614318\_27978884522397  
get value is 1516092614318\_27978884522397  
set value is 1516092614318\_27978884541368  
get value is 1516092614318\_27978884541368

一个生产者和一个消费者进行数据的交互，set和get交替运行。

2， 多个生产者与多个消费者：操作值，可能假死  
测试代码：

public class Producer {
    	private Object lock;
    	public Producer(Object lock){
    		this.lock = lock;
    	}
    	public void setValue(){
    		try {
    			synchronized(lock){
    				while(!ValueObject.value.equals("")){
    					System.out.println("生产者"+Thread.currentThread().getName()+"Waiting ***");
    					lock.wait();
    				}
    				System.out.println("生产者"+Thread.currentThread().getName()+"Running ^^^");
    				String value = System.currentTimeMillis()+"_"+System.nanoTime();
    				System.out.println("set value is "+value);
    				ValueObject.value = value;
    				lock.notify();
    			}
    		} catch (InterruptedException e) {
    		}
    	}
    }

public class Consumer {
    	private Object lock;
    	public Consumer(Object lock){
    		this.lock = lock;
    	}
    	public void getValue(){
    		try {
    			synchronized(lock){
    				while(ValueObject.value.equals("")){
    					System.out.println("消费者 "+Thread.currentThread().getName()+"Waiting ***");
    					lock.wait();
    				}
    				System.out.println("消费者 "+Thread.currentThread().getName()+"Running ^^^");
    				ValueObject.value = "";
    				lock.notify();
    			}
    		} catch (InterruptedException e) {
    		}
    	}
    }

public class RunDemo {
    
    	public static void main(String[] args) throws InterruptedException {
    		Object obj = new Object();
    		Producer p = new Producer(obj);
    		Consumer c = new Consumer(obj);
    		ThreadP[] tp = new ThreadP[2];
    		ThreadC[] tc = new ThreadC[2];
    		for(int i=0; i<2; i++){
    			tp[i] = new ThreadP(p);
    			tp[i].setName("生产者"+(i+1));
    			tc[i] =new ThreadC(c);
    			tc[i].setName("消费者"+(i+1));
    			tp[i].start();
    			tc[i].start();
    		}
    		Thread.sleep(2000);
    		Thread[] tA = new Thread[Thread.currentThread().getThreadGroup().activeCount()];
    		Thread.currentThread().getThreadGroup().enumerate(tA);
    		for(int i=0; i<tA.length; i++){
    			System.out.println(tA[i].getName()+"-"+tA[i].getState());
    		}
    	}
    }

运行结果：

...

消费者 消费者1Waiting \*\*\*  
生产者生产者1Running ^^^  
set value is 1516094855740\_30220204973226  
生产者生产者1Waiting \*\*\*  
生产者生产者2Waiting \*\*\*  
消费者 消费者2Running ^^^  
消费者 消费者2Waiting \*\*\*  
消费者 消费者1Waiting \*\*\*  
main-RUNNABLE  
生产者1-WAITING  
消费者1-WAITING  
生产者2-WAITING  
消费者2-WAITING

呈假死状态的线程都呈WAITING状态，虽然使用了wait/notify，但不能保证nofity唤醒的是异类，也可能是同类，比如生产者唤醒生产者，这种情况累积后，程序最后就是“假死”状态。

将notify改成notifyall可以解决这个“假死”的问题。

另外利用wait/notify机制，可以让多个线程之间交叉执行。