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() {
            @Override
            public 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;
        }

        @Override
        public 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");
        //启动线程printThread
        printThread.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;
        }

        @Override
        public void run() {
            int receive = 0;
            try{
                while ((receive = reader.read()) != -1){
                    //字符转换
                    System.out.print((char) receive);
                }
            }catch (IOException e) {
                System.out.print(e);
            }
        }
    }
}