线程间通信 wait notify

线程间通信

一、通信

线程通信概念：线程是操作系统中独立的个体，但这些个体如果不经过特殊的处理就不能成为一个整体，线程间的通信就成为整体的必用方式之一。当线程存在通信指挥，系统间的交互性会更大，在提高CPU利用率的同事还会使开发人员对线程任务在处理的过程中进行有效的把控与监督。

二、

1.通过 while(true )监听线程

package com.study.current.thread.day02;

import java.util.ArrayList;
import java.util.List;

/**
 * 线程 t1 循环加入元素
 * 线程 t2 监听集合大小，当集合大小为5时，停止线程执行
 * 
 * 当 t2 抛出异常后，但 t1 未停止运行
 *
 */
public class ListAdd1 {

	private volatile static List list = new ArrayList();
	
	public void add (){
		list.add("abcd");
	}
	
	public int size(){
		return list.size();
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {

		final ListAdd1 list1 = new ListAdd1();
		Thread thread1 = new Thread(new Runnable() {
			
			public void run() {
				for(int i = 0 ;i < 10 ; i++){
					list1.add();
					System.out.println(Thread.currentThread().getName() + " list add ");
					try {
						Thread.sleep(500);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}
		},"t1");
		
		Thread thread2 = new Thread(new Runnable() {
			
			public void run() {
				while(true){
					if(list1.size() == 5){
						System.out.println(Thread.currentThread().getName()+ " stop ");
						throw new RuntimeException();
					}
				}
			}
		},"t2");
		
		thread1.start();
		thread2.start();
	}

}

2.使用 wati notify 优化

package com.study.current.thread.day02;

import java.util.ArrayList;
import java.util.List;

/**
 * 线程 t1 循环加入元素
 * 线程 t2 监听集合大小，当集合大小为5时，停止线程执行
 * 
 * 当 t2 抛出异常后，但 t1 未停止运行
 * 
 * 使用 wait notify 
 *
 * 弊端：非实时性的， t1 一直持有 锁
 */
public class ListAdd2 {

	private volatile static List list = new ArrayList();
	
	public void add (){
		list.add("abcd");
	}
	
	public int size(){
		return list.size();
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {

		final ListAdd2 list1 = new ListAdd2();
		
		final Object lock = new Object();
		
		Thread thread1 = new Thread(new Runnable() {
			
			public void run() {
				synchronized (lock) {
					/**
					 * 2.
					 * t1 线程开始执行
					 * 当 size == 5 是，通知 线程2 ，但 notify 不释放锁
					 * 依然是 t1 持有锁，继续执行，知道 synchronized 中的方法体执行完毕
					 */
					for(int i = 0 ;i < 10 ; i++){
						list1.add();
						System.out.println(Thread.currentThread().getName() + " list add ");
						try {
							Thread.sleep(500);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
						if(list1.size() == 5){
							System.out.println("send message to t2 ");
							lock.notify();
						}
					}
				}
			}
		},"t1");
		
		/**
		 * 1.
		 * t2 线程先启动，监控
		 * 当  list1.size != 5 时 ,wait 释放锁
		 * t1 线程可以执行
		 */
		Thread thread2 = new Thread(new Runnable() {
			
			public void run() {
				/**
				 * 替换 ListAdd1 中的 while 监控
				 */
				synchronized (lock) {
					System.out.println("t2 start ");
					
					if(list1.size() != 5){
						
						try {
							lock.wait();
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}
					System.out.println(Thread.currentThread().getName()+ " receive message from t1 ");
					throw new RuntimeException();
				}
				
			}
		},"t2");
		
		thread2.start();
		thread1.start();
	}

}

3.实时性

package com.study.current.thread.day02;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * 线程 t1 循环加入元素
 * 线程 t2 监听集合大小，当集合大小为5时，停止线程执行
 * 
 * 当 t2 抛出异常后，但 t1 未停止运行
 * 
 * 使用 wait notify 
 *
 */
public class ListAdd3 {

	private volatile static List list = new ArrayList();
	
	public void add (){
		list.add("abcd");
	}
	
	public int size(){
		return list.size();
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {

		final ListAdd3 list1 = new ListAdd3();
		
//		final Object lock = new Object();
		// 实时通知
		final CountDownLatch countDownLatch = new CountDownLatch(1);
		Thread thread1 = new Thread(new Runnable() {
			
			public void run() {
//				synchronized (lock) {
					/**
					 * 2.
					 * t1 线程开始执行
					 * 当 size == 5 是，通知 线程2 ，但 notify 不释放锁
					 * 依然是 t1 持有锁，继续执行，知道 synchronized 中的方法体执行完毕
					 */
					for(int i = 0 ;i < 10 ; i++){
						list1.add();
						System.out.println(Thread.currentThread().getName() + " list add ");
						try {
							Thread.sleep(500);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
						if(list1.size() == 5){
							System.out.println("send message to t2 ");
//							lock.notify();
							countDownLatch.countDown();
						}
					}
//				}
			}
		},"t1");
		
		/**
		 * 1.
		 * t2 线程先启动，监控
		 * 当  list1.size != 5 时 ,wait 释放锁
		 * t1 线程可以执行
		 */
		Thread thread2 = new Thread(new Runnable() {
			
			public void run() {
				/**
				 * 替换 ListAdd1 中的 while 监控
				 */
//				synchronized (lock) {
					System.out.println("t2 start ");
					
					if(list1.size() != 5){
						try {
							countDownLatch.await();
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
//						try {
//							lock.wait();
//						} catch (InterruptedException e) {
//							e.printStackTrace();
//						}
					}
					System.out.println(Thread.currentThread().getName()+ " receive message from t1 ");
					throw new RuntimeException();
//				}
				
			}
		},"t2");
		
		thread2.start();
		thread1.start();
	}

}

三、模拟queue

package com.study.current.thread.day02;

import java.util.LinkedList;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 
 * 使用 wait / notify 模拟 Queue 
BlockingQueue：顾名思义，首先它是一个队列，并且支持阻塞的机制，阻塞的放入和得到数据。
put(anObject) 把 anObject 加到 BlockingQueue 里，如果 BlockingQueue 没有空间，则调用此方法的线程被阻断，
直到 BlockingQueue 里面有空间再继续。
take:取走 BlockingQueue 里排在首位的对象，若 BlockingQueue 为空，阻断进入等待状态直到 BlockingQueue 有新的数据被加入
 */
public class MyQueue {
	
	// 定义一个集合
	private final LinkedList<Object> list = new LinkedList<Object>();
	
	// 定义一个计数器，加入list中的元素的数量
	private AtomicInteger count = new AtomicInteger(0);
	
	// 定义集合上下限
	private final int minSize = 0 ;
	private final int maxSize  ; // 创建队列时指定最大长度
	
	// 构造方法，指定集合最大长度
	public MyQueue(int size){
		this.maxSize = size ;
	}
	
	// 定义锁对象
	private Object lock = new Object();
	
	// 放入元素
	public void put(Object anObject){
		synchronized (lock) {
			// 容器已满
			while(maxSize == count.get()){
				try {
					lock.wait();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			// 不满
			list.add(anObject);
			System.out.println("新加入的元素："+anObject);
			// 集合内元素数量加 1
			count.incrementAndGet();
			// 通知另外一个线程，可以取数据了
			lock.notify();
		}
	}
	
	// 取出元素
	public Object take(){
		Object obj = null ;
		synchronized (lock) {
			// 容器为空
			while(count.get() == minSize){
				try {
					lock.wait();
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
			// 不为空
			obj = list.remove();
			System.out.println("取出元素："+obj);
			// 计数递减
			count.decrementAndGet();
			// 通知另外一个线程
			lock.notify();
		}
		return obj ;
	}
	
	public int getSize(){
		return this.count.get();
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		final MyQueue queue = new MyQueue(5);
		queue.put("a");
		queue.put("b");
		queue.put("c");
		queue.put("d");
		queue.put("e");
		
		System.out.println("当前的容器长度："+queue.getSize());
	
		Thread t1 = new Thread(new Runnable() {
			// 线程内使用变量，必须用final 修饰
			public void run() {
				queue.put("f");
				queue.put("g");
			}
		});
		try {
			Thread.currentThread();
			Thread.sleep(2000);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		
		Thread t2 = new Thread(new Runnable() {
			
			public void run() {
				Object obj = queue.take();
				System.out.println("移除的元素为："+obj);
				Object obj1 = queue.take();
				System.out.println("移除的元素为："+obj1);
			}
		});
		
		t1.start();
		t2.start();
	}

}

运行结果：

新加入的元素：a
新加入的元素：b
新加入的元素：c
新加入的元素：d
新加入的元素：e
当前的容器长度：5
取出元素：a
移除的元素为：a
取出元素：b
移除的元素为：b
新加入的元素：f
新加入的元素：g


t1 先运行，插入元素，执行 while ，等待 t2 进行消费
t2 消费完毕，通知 t1 可以插入数据

线程间的通信实现