线程间通信的定义

**定义：**当多个线程共同操作共享的资源时，线程间通过某种方式互相告知自己的状态，以避免无效的资源争夺。

举个栗子：我们在生产中不能闭门造车，比如生产饮料的过程中，有以下流水线:

• 生产可乐瓶；
• 生产饮料
• 装瓶

为了提高生产效率，每一步之间不能割裂进行，当生产的瓶子和饮料足够的时候，这两条生产线告知装瓶的生产线，开始进行正常装瓶。这个过程就可以看做是线程间的通信

通信的方式：

1. 等待—通知
2. 共享内存
3. 管道流
4. ReentrantLock/Condition 消息队列方式

**线程同步：**即当有一个线程在对内存进行操作时，其他线程都不可以对这个内存地址进行操作，直到该线程完成操作， 其他线程才能对该内存地址进行操作。同步的话，仅仅传递的是控制信息，就是我什么时候运行结束，你什么时候可以来。

对于线程间通信来说，线程间同步可以归纳为线程间通信的一个子集，对于线程通信指的是两个线程之间可以交换一些实时的数据信息，而线程同步只交换一些控制信息。

在java中有几个关于线程间通信的关键字：

1.volatile

关键字volatile可以用来修饰字段（成员变量），就是告知程序任何对该变量的访问均需要从共享内存中获取，而对它的改变必须同步刷新回共享内存，它能保证所有线程对变量访问的可见性。

线程会将内存中的数据，拷贝到各自的本地内存中( 这里的本地内存指的是 cpu cache ( 比如 CPU 的一级缓存、二级缓存等 )，寄存器)。当某个变量被 volatile 修饰并且发生改变时，volatile 变量底层会通过 lock前缀的指令，将该变量写回主存，同时利用 缓存一致性协议，促使其他线程的本地变量的数据无效，从而再次直接从主存读取数据。

2.synchronized

关键字synchronized可以修饰方法或者以同步块的形式来进行使用，它主要确保多个线程在同一个时刻，只能有一个线程处于方法或者同步块中，它保证了线程对变量访问的可见性和排他性。

任意一个对象都拥有自己的监视器，当这个对象由同步块或者这个对象的同步方法调用时，执行方法的线程必须先获取到该对象的监视器才能进入同步块或者同步方法，而没有获取到监视器（执行该方法）的线程将会被阻塞在同步块和同步方法的入口处，进入BLOCKED状态。

对象、对象的监视器、同步队列和执行线程之间的关系：

任意线程对Object（Object由synchronized保护）的访问，首先要获得Object的监视器。如果获取失败，线程进入同步队列，线程状态变为BLOCKED。当访问Object的前驱（获得了锁的线程）释放了锁，则该释放操作唤醒阻塞在同步队列中的线程，使其重新尝试对监视器的获取。

一、等待—通知

**等待通知机制wait/notify：**一个线程修改了一个对象的值，二另一个线程感知到了变化，然后进行相应的操作，整个开始与一个线程，而最终执行又是另一个线程。

等待—通知机制使用的是使用同一个对象锁，如果两个线程使用的是不同的对象锁，那它们之间是不能用等待—通知机制的通信的。

（1）等待—通知机制的相关方法：

方法名称	含义
notify()	通知一个对象上等待的线程，使其从wait()方法返回，而返回的前提是该线程取得了对象的锁
notifyAll()	通知所有等待在该对象上的线程
wait()	调用该方法的线程进入WAITING状态，只有等待另外线程的通知或被终端才会返回，调用wait()方法后，会释放对象的锁
wait(long timeout)	超时等待一段时间，没有通知就超时返回，参数时间是毫秒
wait(long timeout, int nanos)	对于超时时间更细粒度的控制，可达到纳秒

（2）注意事项：

• 使用wait()、notify()和notifyAll()时需要先对调用对象加锁；
• 调用wait()方法后，线程状态由RUNNING变为WAITING，并将当前线程放置到对象的等待队列。
• notify()或notifyAll()方法调用后，等待线程依旧不会从wait()返回，需要调用notify()或notifAll()的线程释放锁之后，等待线程才有机会从wait()返回。
• notify()方法将等待队列中的一个等待线程从等待队列中移到同步队列中，而notifyAll()方法则是将等待队列中所有的线程全部移到同步队列，被移动的线程状态由WAITING变为BLOCKED。
• 从wait()方法返回的前提是获得了调用对象的锁

等待/通知机制依托于同步机制，其目的就是确保等待线程从wait()方法返回时能够感知到通知线程对变量做出的修改。在synchronized修饰的同步方法或者修饰的同步代码块中使用Object类提供的wait(),notify()和notifyAll()3个方法进行线程通信。

##（3）wait()方法的核心原理

1）当线程调用了locko（某个同步锁对象）的wait()方法后，JVM会将当前线程加入locko监视器的WaitSet（等待集），等待被其他线程唤醒。

2）当前线程会释放locko对象监视器的Owner权限，让其他线程可以抢夺locko对象的监视器。

3）让当前线程等待，其状态编程WAITING。

在线程调用了同步对象locko的wait()方法之后，同步对象locko的监视器内部状态大致如下图所示：

（4）notify()方法的核心原理

1）当线程调用了locko（某个同步锁对象）的notify()方法后，JVM会唤醒locko监视器WaitSet中的第一条等待线程。

2）当线程调用了locko的notifyAll()方法后，JVM会唤醒locko监视器WaitSet中的所有等待线程。

3）等待线程被唤醒后，会从监视器的WaitSet移动到EntryList，线程具备了排队抢夺监视器Owner权利的资格，其状态从WAITING变成BLOCKED。

在线程调用了同步对象locko的wait()或者notifyAll()方法之后，同步对象locko的监视器内部状态大致如下图所示：

（5）等待—通知机制的经典范式

等待方遵循如下原则。
1）获取对象的锁。
2）如果条件不满足，那么调用对象的wait()方法，被通知后仍要检查条件。
3）条件满足则执行对应的逻辑。

synchronized(对象){while(条件不满足){对象.wait();}//处理逻辑部分
}

通知方遵循如下原则。
1）获得对象的锁。
2）改变条件。
3）通知所有等待在对象上的线程。

synchronized(对象){改变条件；对象.notifyAll();
}

完整代码：

package priv.wdragon.communications;import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.TimeUnit;/*** 等待/通知的经典范式* 等待方遵循如下原则。* 	1）获取对象的锁。* 	2）如果条件不满足，那么调用对象的wait()方法，被通知后仍要检查条件。* 	3）条件满足则执行对应的逻辑。* 	synchronized(对象){* 		while(条件不满足){* 			对象.wait();*        	}* 		//处理逻辑部分* 	}** 通知方遵循如下原则。* 	1）获得对象的锁。* 	2）改变条件。* 	3）通知所有等待在对象上的线程。* 	synchronized(对象){* 		改变条件；* 		对象.notifyAll();* 	}*/public class WaitNotify {static boolean flag = true;static Object lock = new Object();public static void main(String[] args) {Thread waitThread = new Thread(new Wait(), "WaitThread");waitThread.start();try {TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}Thread notifyThread = new Thread(new Notify(), "notifyThread");notifyThread.start();}static class Wait implements Runnable {@Overridepublic void run() {//加锁，拥有lock的Moitorsynchronized (lock) {while (flag) {try {System.out.println(Thread.currentThread() + " flag is true. waitting@" + new SimpleDateFormat("HH:mm:ss").format(new Date()));// Thread.currentThread()可以获取当前线程的引用，// 一般都是在没有线程对象又需要获得线程信息时通过Thread.currentThread()获取当前代码段所在线程的引用// Thread[当前主线程的名字,线程级别,线程组]//    getId();获取该线程的标识符//　　getName();获取该线程名称//　　getState();获取线程状态//　　boolean isAlive();测试线程是否属于活动状态//　　boolean isDaemon();测试线程是否为守护线程//　　boolean isInterrupted();测试线程是否已经中断//执行wait之后会放弃锁并进入对象的等待队列中，进入等待状态//当被唤醒后会自动重新获得锁//而sleep就是直接去睡觉不会释放锁/*线程状态由 RUNNING 变为 WAITING，并将当前线程放置到对象的等待队列中*/lock.wait();} catch (InterruptedException e) {e.printStackTrace();}}System.out.println(Thread.currentThread() + " flag is false. running@" + new SimpleDateFormat("HH:mm:ss").format(new Date()));// new Date()：获取当前时间}}}static class Notify implements Runnable {@Overridepublic void run() {synchronized (lock) {//获取lock的锁，然后进行通知，通知是不会释放lock的锁，//直到当前线程释放了 lock 后，WaitThread才能从 wait 方法中返回System.out.println(Thread.currentThread() + " hold lock. notify@" + new SimpleDateFormat("HH:mm:ss").format(new Date()));/*使用 nofity() 会更加的高效。需要注意的是，nofity() 在某些情况下却会导致死锁，所以只有在经过精细地设计后，才能使用 nofity()。总的来讲，一开始应该总是使用 notifyAll()，只有在发现确实它导致性能问题时，才考虑 notify()，并且对死锁问题给予足够的关注。唤醒并一定真得能立刻唤醒，它需要等待调用 notify()或notifyAll() 的线程释放锁之后，等待线程才有机会从 wait() 返回。notify() 方法将等待队列中的一个等待线程从等待队列中移到同步队列中，而 notifyAll() 方法则是将等待队列中所有线程全部移动到同步对象。被移动的线程状态由 WAITING 变为 BLOCKED*/flag = false;lock.notifyAll();try {TimeUnit.SECONDS.sleep(2);} catch (InterruptedException e) {}}synchronized (lock) {System.out.println(Thread.currentThread() + " hold lock. notify@" + new SimpleDateFormat("HH:mm:ss").format(new Date()));try {//而sleep就是直接去睡觉不会释放锁//所以 lock.wait() 想要 re-obtain ownership of the monitor and resumes execution// 必须等待 睡眠结束TimeUnit.SECONDS.sleep(1);} catch (InterruptedException e) {e.printStackTrace();}}synchronized (lock) {System.out.println(Thread.currentThread() + " hold lock. notify@" + new SimpleDateFormat("HH:mm:ss").format(new Date()));try {//而sleep就是直接去睡觉不会释放锁//所以 lock.wait() 想要 re-obtain ownership of the monitor and resumes execution// 必须等待 睡眠结束TimeUnit.SECONDS.sleep(5);} catch (InterruptedException e) {e.printStackTrace();}}}}
}

执行结果：

（6）Thread.join()

如果一个线程A执行了thread.join()语句，其含义是： 当前线程A等待thread线程终止之后才从thread.join()返回。 线程Thread除了提供join()方法之外，还提供了join(long millis)和join(longmillis, int nanos)两个具备超时特性的方法。这两个超时方法表示，如果线程thread在给定的超时时间里没有终止，那么将会从该超时方法中返回。

创建了10个线程，编号0～9，每个线程调用前一个线程的join()方法 ，也就是线程0结束了，线程1才能从join()方法中返回，而线程0需要等待main线程结束。

package priv.wdragon.communications;import java.util.concurrent.TimeUnit;public class JoinTest {public static void main(String[] args) throws InterruptedException {Thread previous = Thread.currentThread();// 第一次previous是main线程for (int i = 0; i < 10; i++) {// 每个线程用有当前一个线程的引用，需要等待前一个线程终止，才能从等待中返回Thread thread = new Thread(new Domino(previous), String.valueOf(i));thread.start();// 将当前创建的线程Thread赋值给previousprevious = thread;}TimeUnit.SECONDS.sleep(5);System.out.println(Thread.currentThread().getName() + " terminate.");}static class Domino implements Runnable {private Thread thread;public Domino(Thread thread) {this.thread = thread;}@Overridepublic void run() {try {thread.join();// 当前等待上一个线程执行结束} catch (InterruptedException e) {e.printStackTrace();}System.out.println(Thread.currentThread().getName() + " terminate.");}}
}

从上述输出可以看到，每个线程终止的前提是前驱线程的终止，每个线程等待前驱线程终止后，才从join()方法返回，这里涉及了等待/通知机制（等待前驱线程结束，接收前驱线程结束通知）。

二、共享内存

（1）同步——synchronized

关键字synchronized可以修饰方法或者同步块，它主要确保多个线程在同一时刻，只能有一个线程处于方法或者同步块中，它保证了线程对变量访问的可见性和排他性。

这种方式，线程需要不断地去尝试获得锁，如果失败了，再继续尝试。这可能会消耗服务器资源。

**实例代码：**线程A执行完，再让线程B执行，使用对象锁实现

package priv.wdragon.communications;public class Objectlock {private static Object lock = new Object();static class ThreadA implements Runnable {@Overridepublic void run() {synchronized (lock) {for (int i = 0; i < 5; i++) {System.out.println("ThreadA" + i);}}}}static class ThreadB implements Runnable {@Overridepublic void run() {synchronized (lock) {for (int i = 0; i < 5; i++) {System.out.println("ThreadB" + i);}}}}public static void main(String[] args) throws InterruptedException {// 线程A和线程B需要访问同一个对象lock，谁获得锁，谁就先执行，// 这里控制的是让线程A先执行（Thread.sleep(10)为的就是A先获得锁）。// 线程B要等线程A执行完再执行，所以是同步的，这就实现了线程间的通信new Thread(new ThreadA()).start();Thread.sleep(10);// 让线程先睡眠10ms，可以确保A先获得锁System.out.println("-----------------");new Thread(new ThreadB()).start();}
}

执行结果：

（2）信号量——volatile

java支持多个线程同时访问一个对象或者对象的成员变量，由于每个线程可以拥有这个变量的拷贝，所以程序在执行过程中，一个线程看到的变量并不一定是最新的。

关键字volatile可以用来修饰字段（成员变量），就是告知程序对该变量的访问均需要从共享内存中获取，而对它的改变必须同步刷新回共享内存，它能保证所有线程对变量访问的可见性。

volatile能够保证内存的可见性，如果在一个县城里面改变了这个变量的值（该变量是由volatile关键字修饰的），那么其它线程对这种改变是立马可见的。

**实例代码：**线程A输出0，然后线程B输出1，再然后线程A输出2…

package priv.wdragon.communications;public class Signal {private static volatile int signal = 0;static class ThreadA implements Runnable {@Overridepublic void run(){while (signal < 5) {if (signal % 2 == 0) {System.out.println("ThreadA: " + signal);synchronized (this) {signal++;}}}}}static class ThreadB implements Runnable {@Overridepublic void run() {while (signal < 5) {if (signal % 2 == 1){System.out.println("ThreadB: " + signal);synchronized (this){signal++;}}}}}public static void main(String[] args) throws InterruptedException {// volatile 变量需要进⾏原⼦操作。 signal++ 并不是⼀个原⼦操// 作，所以我们需要使⽤ synchronized 给它“上锁”new Thread(new ThreadA()).start();Thread.sleep(10);new Thread(new ThreadB()).start();}
}

执行结果：

关于synchronized与volatile ，synchronized主要做的是多线程顺序执行，也就是同一个时间只有一个线程在执行，线程A执行完了再让线程B执行，volatile主要做的是让多线程间共享的变量保证一致，也就是线程A对变量操作了，线程B对变量操作时是知道线程A对变量的操作的，是在线程A操作后的变量上进行操作。

（3）循环队列：生产者消费者模型

**概念：**生产者消费者模式，是通过一块缓冲区作为容器，来解决生产者和消费者之间的强耦合关系。通俗来讲就是，在该模型之前，只有当前顾客来了，店家才会生产商品，这样的话，来多个顾客就会将大量时间浪费在排队上，白白浪费时间。而有了生产者消费者模式之后，店家在没有顾客的时候，也生产商品，并将商品放在一个容器中，顾客来了直接拿即可，只有当容器满了便停止生产。而生活中大多都采用这个模型。

实例代码：

使用synchronized实现多生产者消费者模型。

package priv.wdragon.communications;/*** 生产者生产物品超过20个就会停一下，消费者消费*/public class ProducerAndConsumer {public static void main(String[] args) {Client client = new Client();Producer producer = new Producer(client);Consumer consumer = new Consumer(client);producer.setName("peoducer1");consumer.setName("consumer1");producer.start();consumer.start();}
}class Client {private int num = 0;public synchronized void product() {if (num < 20) {num++;System.out.println(Thread.currentThread().getName() + " 生产者生产了第 " + num + " 个产品");notify();} else {try {wait();} catch (InterruptedException e) {e.printStackTrace();}}}public synchronized void consume() {if (num > 0) {System.out.println(Thread.currentThread().getName() + " 消费者消费了第" + num + " 个产品");num--;notify();} else {try {wait();} catch (InterruptedException e) {e.printStackTrace();}}}
}class Producer extends Thread {private Client client;public Producer(Client client) {this.client = client;}@Overridepublic void run() {System.out.println(getName() + " 生产者开始生产.......");while (true) {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}client.product();}}
}class Consumer extends Thread {private Client client;public Consumer(Client client) {this.client = client;}@Overridepublic void run() {System.out.println(getName() + " 消费者开始消费.......");while (true) {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}client.consume();}}
}

注意：

• 生产者生产的时候，消费者不能消费；
• 消费者消费的时候，生产者不能生产；
• 容器空了，消费者不能消费，需要通知生产者该干活了；
• 容器满了，生产者不能生产；
• 生产者和消费者共享一个容器，共享数据作为锁。

生产者消费者模型的优点：

• 解耦，通过缓冲区，将生产者与消费者之间进行解耦，谁也不关心谁，只关心容器中的商品数量；
• 提高速率，通过平衡生产者和消费者的处理的处理能力，提高整体的运作能力。

面试题：sleep()和wait()的区别：

• sleep()是Thread类的方法，wait()是Object类中的方法，因此任意对象都可以调用wait()方法，但仅有Thread类对象才能让调用sleep方法；
• sleep()和wait()的功能都是让线程休眠，但sleep()不需要唤醒，wait()需要唤醒；
• 线程在调用sleep()猴，不会释放对象锁，因此其他进程仍然进不了同步块，但wait()调用猴会立即释放对象锁；
• wait()方法必须在同步块或者同步方法中使用，但sleep()没有该要求。

三、 管道流

管道是基于“管道流”的通信方式，管道输入/输出流要用于线程之间的数据传输，而传输的媒介为内存。

管道输入/输出流的体现：

• 基于字符的：PipedWriter、PipedReader
• 基于字节流的：PipedOutputStream、PipedInputStream

使用管道多半与I/O流相关。当我们一个线程需要先向另一个线程发送一个信息（比如字符串）或者文件等等时，就需要使用管道通信了。像消息传递机制，通过管道，将一个线程中的消息发送给另一个。

实例代码：

基于PipedWriter和PipedReader的实现ReaderThread和WriterThread的通信

WriterThread写了内容，ReaderThread读到并打印

package priv.wdragon.communications;import java.io.IOException;
import java.io.PipedReader;
import java.io.PipedWriter;public class Piped {public static void main(String[] args) throws IOException, InterruptedException {PipedWriter pipedWriter = new PipedWriter();PipedReader pipedReader = new PipedReader();pipedWriter.connect(pipedReader);// 注意：// 对于Piped类型的流，必须先要进行绑定，也就是调用connect()方法，// 如果没有将输入/输出流绑定起来，对于该流的访问将会抛出异常。new Thread(new ReaderThread(pipedReader)).start();Thread.sleep(10);new Thread(new WriterThread(pipedWriter)).start();}static class ReaderThread implements Runnable {private PipedReader in;public ReaderThread(PipedReader in){this.in = in;}@Overridepublic void run() {System.out.println("This is a Reader");int receice = 0;try {while ((receice=in.read()) != -1){System.out.println("read " + (char)receice);}} catch (IOException e) {e.printStackTrace();}}}static class WriterThread implements Runnable {private PipedWriter out;public WriterThread(PipedWriter out) {this.out = out;}@Overridepublic void run() {System.out.println("This is a Writer");try {out.write("write A");} catch (IOException e) {e.printStackTrace();}finally {try {out.close();} catch (IOException e) {e.printStackTrace();}}}}
}

执行结果：

四、ReentrantLock/Condition 消息队列方式

ReentrantLock锁的实现原理虽然和synchronized不用，但是它和synchronized一样都是通过保证线程间的互斥访问临界区，来保证线程安全，实现线程间的通信。相比于synchronized使用Object类的三个方法来实现线程的阻塞和运行两个状态的切换，ReentrantLock使用Condition阻塞队列的await()、signal()、signalAll()三个方法来实现线程阻塞和运行两个状态的切换，进而实现线程间的通信。

package priv.wdragon.communications;import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;public class ReentrantLocConditon {public static void main(String[] args) {ReentrantLock lock = new ReentrantLock();Condition condition = lock.newCondition();List<String> list = new ArrayList<>();// 实现线程AThread threadA = new Thread(() -> {lock.lock();for (int i = 1; i <= 10; i++) {list.add("abc");System.out.println("线程A向list中添加一个元素，此时list中的元素个数为：" + list.size());try {Thread.sleep(500);} catch (InterruptedException e) {e.printStackTrace();}if (list.size() == 5)condition.signal();}lock.unlock();});// 实现线程BThread threadB = new Thread(() -> {lock.lock();if (list.size() != 5) {try {condition.await();} catch (InterruptedException e) {e.printStackTrace();}}System.out.println("线程B收到通知，开始执行自己的业务...");lock.unlock();});threadB.start();try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}threadA.start();}
}

执行结果