Java 多线程(三) 线程通信

╰半橙微兮° 2022-03-02 00:58 328阅读 0赞

### 前言 ###

在前一章我们介绍了线程中较为重要的几个关键字`synchronized`与`volatile`.`synchronized`关键字主要是用于标示线程的同步关系与锁.`volatile`主要是用于将线程内的局部变量与进程总变量之间的交互关系.

本文我们将介绍下线程之间的相互通信.本章主要包括如下的几个部分的内容:

*  wait()方法与notify()方法
 *  join()方法
 *  经典的生产者&消费者实现
 *  ThreadLocal类的使用

--------------------

### 正文 ###

##### wait()方法与notify()方法 #####

在说明`wait()与notify()`方法之前,我们先说下`等待/通知机制`.  
有如下的两个线程:

*  A 线程执行 -》线程挂起 -》线程接受通知
 *  B 线程执行 -》线程通知A线程 -》 线程运行结束  
    其实这特别像我们生活中的一些场景.  
    小A: 小B,明天7点叫我起床.然后A睡觉了,也就是进行挂起,进行等待.  
    小B: 做自己的事情,到B点的时候.通知A进行起床.  
    在设计模式内,这是通知者模式的一种反向运用.

OK,我们扯回原来的话题.那么,上述的`等待/通知 机制`在Java内是如何实现的?  
在Java内的锁Lock的概念并未设计出之前,我们是通过`wait()/notify()`的方式进行实现的.这个做法有个缺点,就是无法准确的通知到某个线程,只能大概的通知.

class WaitThread extends Thread{
    	private Object lock;
    	public WaitThread(Object lock){
    		this.lock = lock;
    	}
    	public void run(){
    		System.out.println("Thread:"+Thread.currentThread().getName()+"Wait begin."+System.currentTimeMillis());
    		synchronized (lock) {
    			try {
    				lock.wait();
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    		System.out.println("Thread:"+Thread.currentThread().getName()+"Wait end."+System.currentTimeMillis());
    	}
    }
    class NotifyThread extends Thread{	
    	private Object lock;
    	public NotifyThread(Object lock){
    		this.lock = lock;
    	}
    	public void run(){
    		System.out.println("Thread:"+Thread.currentThread().getName()+"Notify begin."+System.currentTimeMillis());
    		synchronized (lock) {
    			try {
    				lock.notify();
    				Thread.sleep(2000);
    			} catch (InterruptedException e) {
    				e.printStackTrace();
    			}
    		}
    		System.out.println("Thread:"+Thread.currentThread().getName()+"Notify end."+System.currentTimeMillis());
    	}
    }
    
    public class WaitNotifyThread {
    	public static void main(String[] args) {
    		Object lock = new Object();
    		Thread threadA = new WaitThread(lock);
    		Thread threadB = new NotifyThread(lock);
    		threadA.start();
    		threadB.start();
    	}
    }
    
    //Thread:Thread-0Wait begin.1552965862322
    //Thread:Thread-1Notify begin.1552965862322
    //Thread:Thread-1Notify end.1552965864327
    //Thread:Thread-0Wait end.1552965864327

根据上述例子,我们可以知道如下几个特性:

*  两个线程之间要想通信,那么就需要使用同一个对象(类比锁对象)进行依托.像上文中的`Object lock`就是如此.
 *  此外,当线程在运行`wait()`与`notfiy()`的时候,必须要获取线程的锁.
 *  最后,`wait()`与`notify()`方法是`Object`类的接口,也就是任何对象都可以使用`wait()/notify()`方法进行线程的通信.
 *  PS: 根据上面的输出可以得知,当运行`wait()`方法时,线程立即释放当前的对象锁.(也就是wait()方法下方的代码不会再进行执行下去).当运行`notify()`方法时,线程不会立即释放锁,而是等当前的`notify()`的线程执行完成后,再进行释放.

##### Wait / Notify Tips #####

*  当线程`wait()`之后,调用`interrupt()`方法将其关闭时,会导致抛出异常;
 *  `notify()`方法一次只能通知一个线程;`notifyall()`的方法可以通知所有的线程.
 *  可以使用`wait(long)`时间段的时间,来解决死锁和长时间未获得通知的问题.

##### 生产者 / 消费者模式 #####

所谓的`生产者/消费者模式`,简单来说就是一方发送数据,一方接收数据.我们同样可以使用`wait()/notify()`进行模拟实现.

class ConsumerThread extends Thread{
    	private Object lock;
    	public ConsumerThread(Object lock){
    		this.lock = lock;
    	}
    	public void consume(){
    		synchronized (lock) {
    			try {
    				while ("".equals(ProducerConsumerThread.msg)) {
    					lock.wait();
    				}
    				System.out.println("Thread:"+Thread.currentThread().getName()+"Msg "+ ProducerConsumerThread.msg);
    				ProducerConsumerThread.msg = "";
    				lock.notify();
    			} catch (InterruptedException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    		}
    	}
    	public void run(){
    		while(true){
    			consume();
    		}
    	}
    }
    class ProducerThread extends Thread{
    	private Object lock;
    	public ProducerThread(Object lock){
    		this.lock = lock;
    	}
    	public void produce(){
    		synchronized (lock) {
    			try {
    				while (!"".equals(ProducerConsumerThread.msg)) {
    					lock.wait();
    				}
    				System.out.println("Thread:"+Thread.currentThread().getName()+"Msg "+ ProducerConsumerThread.msg);
    				ProducerConsumerThread.msg = "MSG"+System.currentTimeMillis();
    				lock.notify();
    			} catch (InterruptedException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    		}
    	}
    	public void run(){
    		while(true){
    			produce();
    		}
    	}
    }
    
    public class ProducerConsumerThread {
    	public static String msg = "";
    	public static void main(String[] args) {
    		Object lock = new Object();
    		Thread producerThread = new ProducerThread(lock);
    		Thread consumerThread = new ConsumerThread(lock);
    		
    		producerThread.start();
    		try {
    			Thread.sleep(1000);
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    		consumerThread.start();
    
    	}
    }
    //Thread:Thread-0Msg 
    //Thread:Thread-1Msg MSG1552968574301
    //Thread:Thread-0Msg 
    //Thread:Thread-1Msg MSG1552968574301
    //Thread:Thread-0Msg 
    //Thread:Thread-1Msg MSG1552968574301
    //...

在本实例中,我们使用了`String`对象作为了承载消息的载体.值得注意的是,控制锁通信的变量与消息的载体尽量不要使用同一个.

> 多对多消息  
> 当遇到多对多的消息时候,我们通常采用`notifyAll()`方法,唤醒所有的线程.

*  List消息体实现  
    我们还可以通过List消息体进行实现.但是值得注意的是List类型的实现类,无论是`ArrayList类型`还是`LinkedList类型`都是线程不安全的.我们可以使用对象锁,进行处理.  
    这种直接使用`synchronized`关键字的做法是非常低效率的.后面我们在讲解`JDK1.5`的`LinkedBlockedQueue`的时候讲解下使用`RententLock`锁与`CAS机制`来对于这块的部分进行优化.

import java.util.ArrayList;
    import java.util.List;
    
    class ConsumerListThread extends Thread{
    	private Object lock;
    	public ConsumerListThread(Object lock){
    		this.lock = lock;
    	}
    	public void consume(){
    		synchronized (lock) {
    			try {
    				while(ProducerConsumerListThread.list.size() == ProducerConsumerListThread.index){
    					lock.wait();
    				}
    				// 获取对象锁
    				synchronized(ProducerConsumerListThread.list){
    					String message = ProducerConsumerListThread.list.get(ProducerConsumerListThread.index++);
    					System.out.println("Thread:"+Thread.currentThread().getName()+"Consume Msg "+ message);	
    				}
    				lock.notify();
    			} catch (InterruptedException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    		}
    	}
    	public void run(){
    		while(true){
    			consume();
    		}
    	}
    }
    class ProducerListThread extends Thread{
    	private Object lock;
    	public ProducerListThread(Object lock){
    		this.lock = lock;
    	}
    	public void produce(){
    		synchronized (lock) {
    			try {
    				// offer index - 标识末尾 说明
    				while(ProducerConsumerListThread.list.size() != ProducerConsumerListThread.index){
    					lock.wait();
    				}
    				// 获取对象锁
    				synchronized(ProducerConsumerListThread.list){
    					String message = "MSG:"+System.currentTimeMillis();
    					ProducerConsumerListThread.list.add(message);
    					System.out.println("Thread: "+Thread.currentThread().getName()+"Produce Msg "+ message);	
    				}
    				lock.notify();
    			} catch (InterruptedException e) {
    				// TODO Auto-generated catch block
    				e.printStackTrace();
    			}
    		}
    	}
    	public void run(){
    		while(true){
    			produce();
    		}
    	}
    }
    
    public class ProducerConsumerListThread {
    	public static List<String> list = new ArrayList<>();
    	public static int index = 0; 
    	public static void main(String[] args) {
    		Object lock = new Object();
    		Thread producerThread = new ProducerListThread(lock);
    		Thread consumerThread = new ConsumerListThread(lock);
    		
    		producerThread.start();
    		try {
    			Thread.sleep(1000);
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    		consumerThread.start();
    
    	}
    }
    //Thread: Thread-0Produce Msg MSG:1552977964484
    //Thread:Thread-1Consume Msg MSG:1552977964484
    //Thread: Thread-0Produce Msg MSG:1552977964484
    //Thread:Thread-1Consume Msg MSG:1552977964484
    //Thread: Thread-0Produce Msg MSG:1552977964484
    //Thread:Thread-1Consume Msg MSG:1552977964484

输出的结果如上所示.

##### join()方法 #####

在现实生活中,我们通常需要等待某样东西的结果再进行运算.比如:医生需要对于病人的病情诊断前,通常需要得到病人的检测结果(检测A/检测B/检测C).这种场景使用Java的`join()`方法即可实现.(当然在JDK1.5之后,JUC包中还提供了`CountDownLaunch`/`CyclicBarrier`/`Semaphore`类优化以及简便完成了这部分的内容.这个我们在后续的章节中会进行讲解.)

class SonThread extends Thread{
    	public void run(){
    		for(int i=0;i<10;i++){
    			synchronized(JoinThread.num){
    				JoinThread.num = JoinThread.num+1;
    				// System.out.println(JoinThread.num);
    			}
    		}
    	}
    }
    public class JoinThread{
    	public static Integer num=0;
    	public static void main(String []args){
    		Thread threadA = new  SonThread();
    		Thread threadB = new SonThread();
    		try {
    			threadA.start();
    			threadB.start();
    			threadA.join();
    			threadB.join();
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    		
    		System.out.println("Num = "+num);
    	}
    }
    // Num = 20

*  `join()`方法可以使当前线程在 `join()`方法完成后再进行执行.
 *  与`wait()`方法类似,`join()`也有`join(long)`这样的接口.防止等待的时间过长,或者发生死锁的情况.
 *  与`wait()`方法类似, 在当前线程`join()`时候,被`interrupt()`中断时.也会抛出`InterruptedException`异常.
 *  `join()`与`wait()`方法一致,在方法运行后,都会释放当前线程所包含的锁.`sleep()`方法是不会释放锁的.(这在前文已经有所提及了.)
 *  在使用`join(long)`的接口的时候,需要特别注意锁的释放与争抢的问题.

public final synchronized void join(long millis) 
        throws InterruptedException {
        long base = System.currentTimeMillis();
        long now = 0;
    
        if (millis < 0) {
                throw new IllegalArgumentException("timeout value is negative");
        }
    
        if (millis == 0) {
            while (isAlive()) {
            wait(0);
            }
        } else {
            while (isAlive()) {
            long delay = millis - now;
            if (delay <= 0) {
                break;
            }
            wait(delay);
            now = System.currentTimeMillis() - base;
            }
        }
        }

[多线程中的join方法——源码分析][join]

##### ThreadLocal类的使用 #####

根据第二章可以知道,当我们访问同一个变量时候,通常需要加入`synchronized`关键字以获取锁.但是,如果我们需要在每个线程内都具有自己独自的变量的时候,应当如何解决呢? 不用担心,Java为我们提供了`ThreadLocal`对象以用于存储线程变量.

`ThreadLocal`类的使用非常简单,我们可以将其当作一个集合.对于ThreadLocal有时操作集合对象,有时操作普通对象.其实例皆如下所示:

import java.util.HashMap;
    
    class ThreadLocalThread extends Thread{
    	public void run(){
    		System.out.println("Thread:"+Thread.currentThread().getName()+"Begin Set "+ ThreadLocalDemo.threadlocalInteger.get());	
    		
    		// set
    		ThreadLocalDemo.threadlocalInteger.set(456);
    		System.out.println("Thread:"+Thread.currentThread().getName()+"After Set "+ ThreadLocalDemo.threadlocalInteger.get());	
    
    		
    		// set2
    		ThreadLocalDemo.threadlocalHashMap.set(new HashMap());
    		ThreadLocalDemo.threadlocalHashMap.get().put("hello", "sonThread");
    		System.out.println("Thread:"+Thread.currentThread().getName()+"Hash Map After Set "+ ThreadLocalDemo.threadlocalHashMap.get().get("hello"));	
    	}
    }
    public class ThreadLocalDemo {
    	// 1. 普通对象
    	public static ThreadLocal<Integer> threadlocalInteger = new ThreadLocal<>();
    	
    	// 2. 集合对象
    	public static ThreadLocal<HashMap> threadlocalHashMap = new ThreadLocal<>();
    
    	public static void main(String[] args) {
    		threadlocalInteger.set(123);
    		threadlocalHashMap.set(new HashMap());
    		threadlocalHashMap.get().put("hello", "main");
    		
    		ThreadLocalThread sonThread = new ThreadLocalThread();
    		
    		try {
    			Thread.sleep(2000);
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    		
    		sonThread.start();
    		System.out.println("Thread:"+Thread.currentThread().getName()+"After Set "+ ThreadLocalDemo.threadlocalInteger.get());	
    		
    		System.out.println("Thread:"+Thread.currentThread().getName()+"Hash Map After Set "+ ThreadLocalDemo.threadlocalHashMap.get().get("hello"));	
    
    	}
    }
    //Thread:mainAfter Set 123
    //Thread:Thread-0Begin Set null
    //Thread:mainHash Map After Set main
    //Thread:Thread-0After Set 456
    //Thread:Thread-0Hash Map After Set sonThread

预期结果如上所示.可以看到,变量在两个线程内各不相同.对于不论是简单数据类型还是集合类型都是如此.

*  ThreadLocal的自动初始化(重写initialValuele类)  
    不知道大家是否注意到上述的输出中的这句话:`Thread:Thread-0Begin Set null`.这是因为ThreadLocal在进行初始化操作的时候是`null`的.我们在使用之前,需要先进行赋值.  
    解决办法1.使用之前,需要先进行赋值. 2.重写ThreadLocal的初始化方法.(个人不是特别推荐 因为后期有可能维护稍微有点困难.)

class ThreadLocalExt extends ThreadLocal{
    	protected Object initialValue(){
    		return "First Object";
    	}
    }

*  继承类 InheritableThreadLocal

import java.util.HashMap;
    
    class InheritThreadLocalThread extends Thread{
    	public void run(){
    		System.out.println("Thread:"+Thread.currentThread().getName()+"Parent Set "+ InheritThreadLocalDemo.threadlocalInteger.get());	
    		InheritThreadLocalDemo.threadlocalInteger.set(77777);
    		System.out.println("Thread:"+Thread.currentThread().getName()+"After Son Set "+ InheritThreadLocalDemo.threadlocalInteger.get());	
    
    	}
    }
    public class InheritThreadLocalDemo {
    	// 1. 普通对象
    	public static InheritableThreadLocal<Integer> threadlocalInteger = new InheritableThreadLocal<>();
    	public static void main(String[] args) {
    		threadlocalInteger.set(4396);
    		Thread sonThread = new InheritThreadLocalThread();
    		try {
    			Thread.sleep(2000);
    			sonThread.start();
    			sonThread.join();
    		} catch (InterruptedException e) {
    			e.printStackTrace();
    		}
    
    		System.out.println("Thread:"+Thread.currentThread().getName()+"Final"+ InheritThreadLocalDemo.threadlocalInteger.get());	
    
    	}
    
    }
    //Thread:Thread-0Parent Set 4396
    //Thread:Thread-0After Son Set 77777
    //Thread:mainFinal4396
    
    //Thread:Thread-0Parent Set 4396
    //Thread:Thread-0After Son Set 77777
    //Thread:Thread-0Parent Set parentThread
    //Thread:Thread-0After Son Set sonThread
    //Thread:mainFinal4396
    //Thread:mainParent Final Set sonThread

由上述可以得知: 1. 子线程继承父线程的`InheritableThreadLocal`中的值. 2. 子线程拥有自己的独立空间, 子线程更改后,父线程并不会受到影响. 3. 对于集合类型来说,父和子其实使用的是同一块空间.(个人理解这其实是和软拷贝有关系.)

继承后的同时,可以重写`protected Object childValue(Object parentValue)` 方法进行自定义配置.但个人不是特别建议这么使用.因为这会破坏使用的运行逻辑.

*  注: ThreadLocal其实维护的是`ThreadName - Value`的一个HashMap,因为线程不具有重名的特性,所以并不会出现线程冲突的情况.

--------------------

### Tips ###

*  java.lang.IllegalMonitorStateException  
    在使用`wait()`方法内获取对象的锁 / 当前线程无法获取某个对象的锁  
    [并发 错误 java.lang.IllegalMonitorStateException: current thread not owner 分析][_ java.lang.IllegalMonitorStateException_ current thread not owner]

--------------------

### Reference ###

\[1\]. [多线程中的join方法——源码分析][join]  
\[2\]. [一个优雅的threadLocal工具类][threadLocal]  
\[3\]. Java 多线程编程核心技术  
\[4\]. Java并发编程的艺术

[join]: https://blog.csdn.net/xdzhouxin/article/details/81225707
[_ java.lang.IllegalMonitorStateException_ current thread not owner]: https://blog.csdn.net/shiziaishuijiao/article/details/25537775
[threadLocal]: https://blog.csdn.net/GFJ0814/article/details/54175550