多线程生命周期与通信（二）通信

线程自启动时，就拥有了自己的栈空间。然后会一直运行直到结束。多线程的目的是多条线程执行不同的逻辑业务从而能够提升业务整体的响应速度，如果线程仅仅是孤零零的执行，不同的逻辑业务就不能最终汇聚成一个完整的业务那么多线程也就失去了意义，这就是为什么要有线程间通信的存在。实现线程之间的通信有以下方法：

一、等待/通知机制

1、介绍

一个线程修改一个对象的值，另一个线程感知变化。线程A调用对象O的wait()进入等待状态，另一个线程B调用对象O的notify()或者 notifuAll()方法，线程A 收到通知后从对象O 的wait()方法返回，执行后续操作。两个线程通过对象O来完成交互，而对象上的wait()和notify/notifyAll()的关系就如同开关信号一样，用来完成等待方和通知方之间的交互工作。

（1）wait();当前线程暂停，等待notify()来唤醒(释放资源)。

（2）使用锁对象的notify()方法可以将正在等待的线程唤醒，但是同时有多个线程都处于等待状态，notify()只是随机唤醒一个。
注：唤醒后的进程进入就绪态，而不是进入运行态。虽然线程被唤醒，但只有当前线程放弃对同步锁对象的锁定，被唤醒的线程才可能执行被执行

（3）notifyAll()
唤醒在此同步锁对象上等待的所有线程。同上，只有当前线程放弃对同步锁对象的锁定，才可能执行被唤醒的线程。

2、规则

2.1、等待方遵循原则

（1）获取对象的锁。

（2）如果条件不满足，那么调用对象的 wait()方法，被通知后仍要检查条件。

（3）条件满足则执行对应的逻辑。

对应的伪代码如下:

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

2.2、通知方遵循原则

（1）获得对象的锁。

（2）改变条件

（3）通知所有等待在对象上的线程。

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

3、demo：

public class Consume {

    private static final Logger logger = LoggerFactory.getLogger(Consume.class);

    private final Object lockValue;

    public Consume(Object object) {
        this.lockValue = object;
    }

    /**
     * 生产者赋值
     */
    public void getValue() {
        synchronized (lockValue) {
            if (ObjectUtils.isEmpty(ProductConsumeValue.value)) {
                try {
                    lockValue.wait();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            logger.info("Consume :{}", ProductConsumeValue.value);
            ProductConsumeValue.value = "";
            lockValue.notifyAll();
        }
    }
}

public class Product {

    private static final Logger logger = LoggerFactory.getLogger(Consume.class);

    private Object lockValue;

    public Product(Object lockValue) {
        this.lockValue = lockValue;
    }

    /**
     * 生产者赋值
     */
    public void setValue() {
        synchronized (lockValue) {
            if (!ObjectUtils.isEmpty(ProductConsumeValue.value)) {
                try {
                    lockValue.wait();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            ProductConsumeValue.value = System.currentTimeMillis() + "_" + System.nanoTime();
            logger.info("Product :{}", ProductConsumeValue.value);
            lockValue.notify();
        }
    }
}

public static void main(String[] args) {
    String value = "";
    Product product = new Product(value);
    Consume consume = new Consume(value);
    ProductThread productThread = new ProductThread(product);
    ConsumerThread consumerThread = new ConsumerThread(consume);
    productThread.start();
    consumerThread.start();
}

二、等待超时模式

1、介绍

调用一个方法时等待一段时间（一般来说是给定一个时间段），如果该方法能够在给定的时间段之内得到结果，那么将结果立刻返回，反之，超时返回默认结果。等待超时模式就是在等待/通知范式基础上增加了超时控制，这使得该模式相比原有范式更具有灵活性，因为即使方法执行时间过长，也不会“永久”阻塞调用者，而是会按照调用者的要求“按时”返回。

2、实现

等待/通知的经典范式，即加锁、条件循环和处理逻辑3个步骤，而这种范式无法做到超时等待。超时等待的加入，在等待通知范式上做出改动：假设超时时间段是T，那么可以推断出在当前时间now+T之后就会超时定义如下变量：等待持续时间：REMAINING=T；超时时间：FUTURE=now+T。这时仅需要wait(REMAINING)即可，在wait(REMAINING)返回之后会将执行：REMAINING=FUTURE–now。如果REMAINING小于等于0，表示已经超时，直接退出，否则将继续执行wait(REMAINING)。

三、管道输入/输出流

1、介绍

管道输入/输出流和普通的文件输入/输出流或者网络输入/输出流不同之处在于，它主要用于线程之间的数据传输，而传输的媒介为内存。管道输入/输出流主要包括了如下4 种具体实现：PipedOutputStream、PipedInputStream、PipedReader 和 PipedWriter，前两种面向字节，而后两种面向字符。

2、demo

public class Piped {

    public static void main(String[] args) throws Exception {

        PipedWriter out = new PipedWriter();
        PipedReader in = new PipedReader();
        // 将输出流和输入流进行连接，否则在使用时会抛出IOException
        out.connect(in);
        Thread printThread = new Thread(new Print(in), "PrintThread");
        printThread.start();
        int receive = 0;
        try {
            while ((receive = System.in.read()) != -1) {
               out.write(receive);
            }

        } finally {
            out.close();
        }
    }

 

    static class Print implements Runnable {
        private PipedReader in;
        public Print(PipedReader in) {
            this.in = in;
        }


        public void run() {
            int receive = 0;
            try {
                while ((receive = in.read()) != -1) {
                    System.out.print((char) receive);
                }

            } catch (IOException ex) {

            }

        }

    }

从system.in即控制台中读入一个字符，转为int，将int值写入到管道输出流中，既然输入流与输出流已经连接，那么在输入流就会读取到int值，转为char就能输出。对于Piped类型的流，必须先进行绑定，也就是调用connect（）方法，如果没有将输入/输出流绑定起来，对于该流的访问将会抛出异常。

四、Thread.join()

1、介绍

当前线程暂停，等待加入的线程运行结束，当前线程继续执行。

如果主线程处理完其他的事务后，需要用到子线程的处理结果，也就是主线程需要等待子线程执行完成之后再结束，这个时候就要用到join()方法了。如果一个线程 A 执行了 thread.join()语句，其含义是：当前线程 A 等待 thread 线程终止之后才从 thread.join()返回。

2、原理

查看源码可以看到底层与等待/通知机制范式一致，即加锁、循环、处理逻辑三个步骤。

3、demo

public class ThreadJoinTest {

    private static final Logger logger = LoggerFactory.getLogger(ThreadJoinTest.class);

    /**
     * 如果主线程处理完其他的事务后，需要用到子线程的处理结果，也就是主线程需要等待子线程执行完成之后再结束，这个时候就要用到 join() 方法了。
     *
     * @param args args
     * @throws InterruptedException 中断异常
     */
    public static void main(String[] args) throws InterruptedException {
        Thread currentThread = Thread.currentThread();
        for (int i = 0; i < 10; i++) {
            JoinThreadTest joinTestTread = new JoinThreadTest(currentThread);
            Thread thread = new Thread(joinTestTread, "线程 " + i);
            thread.start();
            currentThread = thread;
        }
        Thread.sleep(5000);
    }

    private static class JoinThreadTest implements Runnable {
        private final Thread thread;

        private JoinThreadTest(Thread currentThread) {
            thread = currentThread;
        }

        @Override
        public void run() {
            try {
                thread.join();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            logger.info("当前线程：{}", Thread.currentThread().getName());
        }
    }
}