Java - 线程间的通信方式

线程通信的方式

线程中通信是指多个线程之间通过某种机制进行协调和交互

线程通信主要可以分为三种方式，分别为共享内存、消息传递和管道流。每种方式有不同的方法来实现

共享内存：线程之间共享程序的公共状态，线程之间通过读-写内存中的公共状态来隐式通信。
volatile共享内存

消息传递：线程之间没有公共的状态，线程之间必须通过明确的发送信息来显示的进行通信。
wait/notify等待通知方式
join方式

管道流
管道输入/输出流的形式

共享内存

/*** @Author: Simon Lang* @Date: 2020/5/5 15:13*/
public class TestVolatile {private static volatile boolean flag=true;public static void main(String[] args){new Thread(new Runnable() {public void run() {while (true){if(flag){System.out.println("线程A");flag=false;}}}}).start();

new Thread(new Runnable() {public void run() {while (true){if(!flag){System.out.println("线程B");flag=true;}}}}).start();}
}

测试结果：线程A和线程B交替执行

消息传递-线程等待和通知

线程等待和通知机制是线程通讯的主要手段之一。

在 Java 中有以下三种实现线程等待的手段：

Object 类提供的 wait()，notify() 和 notifyAll() 方法；
Condition 类下的 await()，signal() 和 signalAll() 方法；
LockSupport 类下的 park() 和 unpark() 方法。

Object 类提供的 wait()，notify() 和 notifyAll() 方法；

Object lock = new Object();
new Thread(() -> {synchronized (lock) {try {System.out.println("线程1 -> 进入等待");lock.wait();System.out.println("线程1 -> 继续执行");} catch (InterruptedException e) {e.printStackTrace();}System.out.println("线程1 -> 执行完成");}
}).start();Thread.sleep(1000);
synchronized (lock) {// 唤醒线程System.out.println("执行 notifyAll()");lock.notifyAll();
}

Condition 类下的 await()，signal() 和 signalAll() 方法；
// 创建 Condition 对象 (lock 可创建多个 condition 对象)
Lock lock = new ReentrantLock();
Condition condition = lock.newCondition();
// 加锁
lock.lock();
try {// 一个线程中执行 await()condition.await();// 另一个线程中执行 signal()condition.signal();
} catch (InterruptedException e) {e.printStackTrace();
} finally {lock.unlock();
}
Condition 类它可以创建出多个对象。那为什么有了 Object 类的 wait 和 notify 的方式，还需要 condition 来干嘛呢？

因为 Object 类的 wait 和 notify 只适用于一个任务队列，而 Condition 类的 await 和 signal 适用于多个任务队列，在多个任务队列的情况下，使用 Object 类的 wait 和 notify 可能会存在线程饿死的问题。

比如以上这种生产者消费者模型，当生产者，消费者（阻塞式的）都有多个的时候，并且此时任务队列里面没有任务了，所以消费者就会进入休眠状态，此时生产者需要做两件事情：

将任务推送到任务队列
唤醒线程
【问题所在】

① 此时如果使用 Object 类提供的 wait 和 notify，而唤醒线程是存在两种可能的：

1）唤醒了消费者

2）唤醒了生产者

如果是唤醒了生产者，那就出问题了，当生产者这边代码执行完了就结束了，消费者这边永远不会去消费队列里的任务了，这就会导致线程饥饿问题。
而 Condition 类因为可以被创建多个，所以可以使用两个 Condition 对象，一个指定唤醒生产者，一个指定唤醒消费者，这样就不会出现线程饥饿了。

所以 Condition 类的 await 和 signal 是对 Object 类的 wait 和 notify 的一个补充，它解决了 Object 类种分组不明确的问题。

LockSupport 类下的 park() 和 unpark() 方法。
public static void main(String[] args) {Thread t1 = new Thread(() -> {LockSupport.park();System.out.println("线程1被唤醒");},"线程1");t1.start();Thread t2 = new Thread(() -> {LockSupport.park();System.out.println("线程2被唤醒");},"线程2");t2.start();Thread t3 = new Thread(() -> {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}System.out.println("唤醒线程2");LockSupport.unpark(t2);},"线程3");t3.start();
}
LockSupport 类又是对 Condition 类的一个补充，它可以指定唤醒某一个线程，它解决了前两种方式不能随机指定唤醒线程的问题。

join方式

join()方法的作用是：在当前线程A调用线程B的join()方法后，会让当前线程A阻塞，直到线程B的逻辑执行完成，A线程才会解除阻塞，然后继续执行自己的业务逻辑，这样做可以节省计算机中资源。

public class TestJoin {public static void main(String[] args){Thread thread=new Thread(new Runnable() {@Overridepublic void run() {System.out.println("线程0开始执行了");}});thread.start();for (int i=0;i<10;i++){JoinThread jt=new JoinThread(thread,i);jt.start();thread=jt;}
}
static class JoinThread extends Thread{private Thread thread;private int i;
public JoinThread(Thread thread,int i){this.thread=thread;this.i=i;}
@Overridepublic void run() {try {thread.join();System.out.println("线程"+(i+1)+"执行了");} catch (InterruptedException e) {e.printStackTrace();}}}
}

每个线程的终止的前提是前驱线程的终止，每个线程等待前驱线程终止后，才从join方法返回，实际上，这里涉及了等待/通知机制，即下一个线程的执行需要接受前驱线程结束的通知。

管道输入/输出流

管道流是是一种使用比较少的线程间通信方式，管道输入/输出流和普通文件输入/输出流或者网络输出/输出流不同之处在于，它主要用于线程之间的数据传输，传输的媒介为管道。

管道输入/输出流主要包括4种具体的实现：PipedOutputStrean、PipedInputStrean、PipedReader和PipedWriter，前两种面向字节，后两种面向字符。

java的管道的输入和输出实际上使用的是一个循环缓冲数组来实现的，默认为1024，输入流从这个数组中读取数据，输出流从这个数组中写入数据，当这个缓冲数组已满的时候，输出流所在的线程就会被阻塞，当向这个缓冲数组为空时，输入流所在的线程就会被阻塞。

public class TestPip {public static void main(String[] args) throws IOException {PipedWriter writer  = new PipedWriter();PipedReader reader = new PipedReader();//使用connect方法将输入流和输出流连接起来writer.connect(reader);Thread printThread = new Thread(new Print(reader) , "PrintThread");//启动线程printThreadprintThread.start();int receive = 0;try{//读取输入的内容while((receive = System.in.read()) != -1){writer.write(receive);}}finally {writer.close();}}
private static class Print implements Runnable {private PipedReader reader;
public Print(PipedReader reader) {this.reader = reader;}
@Overridepublic void run() {int receive = 0;try{while ((receive = reader.read()) != -1){//字符转换System.out.print((char) receive);}}catch (IOException e) {System.out.print(e);}}}
}