Java多线程——线程间通信之wait(),notify()方法,生产者与消费者模式实现,管道流

1.  wait()方法：使当前执行代码的线程进行等待。

        在调用wait()之前，线程必须获得该对象的对象级别锁，即只能在同步方法或同步块中调用wait()方法（否则会抛出IllegalMonitorStateException异常）。在执行wait()方法后，当前线程释放锁。在从wait()返回之前，线程与其他线程竞争重新获得锁。

package threadCommunication;

public class WaitTest {

public static void main(String[] args) {
try {
String str = new String();
System.out.println("synchronize前");
synchronized (str) {
System.out.println("wait前");
str.wait();
System.out.println("wait后");
}
System.out.println("synchronize后");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

输出：（程序没有执行结束，线程一直处于等待状态）

synchronize前
wait前

2.  notify()方法：使停止的线程继续运行。

        在调用notify()之前，线程必须获得该对象的对象级别锁，即只能在同步方法或同步块中调用notify()方法（否则会抛出IllegalMonitorStateException异常）。

        notify()方法用来通知那些可能等待该对象的对象锁的其他线程，如果有多个线程等待，则有线程规划器随机挑选出一个呈wait()状态的线程，对其发出通知notify，并使它等待获取该对象的对象锁。

        在执行notify()方法后，当前线程不会马上释放该对象锁，呈wait()状态的线程也并不能马上获取该对象锁，要等到执行notify()方法的线程将程序执行完，也就是退出synchronized代码后，当前线程才会释放锁，而呈wait()状态所在的线程才可以获取该对象锁。

package threadCommunication;

/**
* 初试notify()方法
*/
public class NotifyTest {
public static void main(String[] args) throws InterruptedException {
Object lock = new Object();
MyThreadA a = new MyThreadA(lock);
a.setName("A");
a.start();
Thread.sleep(2000);
MyThreadB b = new MyThreadB(lock);
b.setName("B");
b.start();
}

}

// 线程A
class MyThreadA extends Thread {
private Object lock;

public MyThreadA(Object lock) {
this.lock = lock;
}

@Override
public void run() {
try {
synchronized (lock) {
System.out.println("线程" + Thread.currentThread().getName() + "开始\t" + System.currentTimeMillis());
lock.wait();
System.out.println("线程" + Thread.currentThread().getName() + "结束\t" + System.currentTimeMillis());
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

//线程B
class MyThreadB extends Thread {
private Object lock;

public MyThreadB(Object lock) {
this.lock = lock;
}

@Override
public void run() {
synchronized (lock) {
System.out.println("线程" + Thread.currentThread().getName() + "开始\t" + System.currentTimeMillis());
lock.notify();
System.out.println("线程" + Thread.currentThread().getName() + "结束\t" + System.currentTimeMillis());

}
}
}

      输出：

线程：A开始	1527514984721
线程：B开始	1527514986723
线程：B结束	1527514986724
线程：A结束	1527514986724

Process finished with exit code 0

3.  notifyall()方法：可以使所有正在等待队列中等待同一共享资源的“全部”线程从等待状态退出，进入可运行状态。此时，优秀级最高的那个线程最先执行，但也有可能是随机执行，这取决于JVM虚拟机的实现。

        一个notify()方法可以唤醒一个线程，当等待同一锁对象的线程较少时，也可以多次（n次）调用notify()唤醒多个（n个）线程。但为了确保能够唤醒所有的线程，可以使用notify()方法。

4.  sleep()方法：线程进入休眠（阻塞）状态，自动放弃当前CPU资源；但是不释放锁。

5.  小总结：

（1）执行完同步代码块就会释放对象的锁。

（2）在执行同步代码块的过程中，遇到异常而导致线程终止，锁也会被释放。

（3）在执行同步代码块的过程中，执行了锁所属对象的wait()方法，这个线程会释放对象锁，而此线程对象会进入线程等待池中，等待被唤醒。

（4）当线程呈wait()状态时，调用线程对象的interrupt()方法会出现InterruptExceptin异常。

6.  wait(long timeout)方法：等待某一时间内是否有其他线程对锁进行唤醒，如果超过了这个时间则自动唤醒。也可以在timeout时间内由其他线程唤醒。

7.  要防止notify通知过早；通知过早会导致wait()进入无限等待，无法唤醒。

8.  所有的类都继承了Object 类。Object类的9个方法：

9.  生产者与消费者模式实现

        等待/通知（wait/notify）模式最经典的案例就是“生产者与消费者”模式。        

(1)一生产与一消费：操作值    

        如果操作值value为null时，customer线程进入wait；product线程执行，给value赋值，执行结束之后，唤醒customer线程并释放对象锁。

        如果操作值value不为null时，product线程进入wait；customer线程执行，获取value后将其置为null，执行结束之后，唤醒product线程并释放对象锁。

package threadCommunication;

/**
* 经典wait/notify模式：生产者与消费者模式
*/
public class ProductAndCustomer {
public static void main(String[] args) throws InterruptedException {
String lock = new String("");
Product pro = new Product(lock);
Customer cus = new Customer(lock);
ThreadP p = new ThreadP(pro);
ThreadC c = new ThreadC(cus);
p.start();
c.start();

}
}

/**
* 生产者
*/
class Product {
private String lock;

public Product(String lock) {
this.lock = lock;
}

public void setValue() {
try {
synchronized (lock) {
if (!ValueObject.value.equals("")) {
lock.wait();// value不为null时，进入等待状态
}
// value为null时，进入生产模式
String value = System.currentTimeMillis() + "_" + System.nanoTime();
System.out.println("set的值是：" + value);
ValueObject.value = value;
lock.notify();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

/**
* 消费者
*/
class Customer {
private String lock;

public Customer(String lock) {
this.lock = lock;
}

public void getValue() {
try {
synchronized (lock) {
if (ValueObject.value.equals("")) {
lock.wait();
}
System.out.println("get的值是：" + ValueObject.value);
ValueObject.value = "";//get之后要将value置空，以备下次生产
lock.notify();
}
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}

/**
* 生产者线程
*/
class ThreadP extends Thread {
private Product pro;// 通过构造方法初始化

public ThreadP(Product pro) {
this.pro = pro;
}

@Override
public void run() {
super.run();
while (true) {
pro.setValue();
}
}
}

/**
* 消费者线程
*/
class ThreadC extends Thread {
private Customer cus;

public ThreadC(Customer cus) {
this.cus = cus;
}

@Override
public void run() {
super.run();
while (true) {
cus.getValue();
}
}
}

/**
* 存储值的工具类
*/
class ValueObject {
public static String value = "";
}

（2）多生产多消费：操作值-假死

        因为多生产多消费，会存在“生产者”唤醒“生产者”，“消费者”唤醒“消费者”这种情况，此情况会导致所有的生产者和消费者都呈waiting状态，线程最后也就呈“假死”状态，不能继续运行下去。

        解决方法：可以使用notifyAl()方法代替notify()方法，一次性唤醒所有的线程。

（3）一生产与一消费：操作栈

        本示例是使生产者向堆栈List对象中放入数据，使消费者从List对象中取出数据。List最大容量为1 。

package threadCommunication;

import java.util.ArrayList;
import java.util.List;
import java.util.Stack;

/**
* (3)一生产与一消费：操作栈
* (4)一生产与多消费：操作栈，解决wait条件改变与假死
*/
public class ProductAndCustomer2 {

private List<String> list = new ArrayList<>();

synchronized public void push() {
//        if (list.size() == 1) {
while (list.size() == 1) {
System.out.println("push的操作：" + Thread.currentThread().getName() + "线程呈wait状态");
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
list.add("anything=" + Math.random());
//        this.notify();
this.notifyAll();
System.out.println("push=" + list.size());
}

synchronized public String pop() {
//        if (list.size() == 1) {
while (list.size() == 0) {
System.out.println("pop的操作：" + Thread.currentThread().getName() + "线程呈wait状态");
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
String value = list.get(0);
list.remove(0);
//        this.notify();
this.notifyAll();
System.out.println("pop=" + list.size());
return value;
}

public static void main(String[] args) {
ProductAndCustomer2 productAndCustomer2 = new ProductAndCustomer2();
P p = new P(productAndCustomer2);
P p2 = new P(productAndCustomer2);
P p3 = new P(productAndCustomer2);
P p4 = new P(productAndCustomer2);
P p5 = new P(productAndCustomer2);
C c = new C(productAndCustomer2);
C c2 = new C(productAndCustomer2);
C c3 = new C(productAndCustomer2);
C c4 = new C(productAndCustomer2);
C c5 = new C(productAndCustomer2);
P_Thread p_thread = new P_Thread(p);
P_Thread p2_thread = new P_Thread(p);
P_Thread p3_thread = new P_Thread(p);
P_Thread p4_thread = new P_Thread(p);
P_Thread p5_thread = new P_Thread(p);
C_Thread c_thread = new C_Thread(c);
C_Thread c2_thread = new C_Thread(c);
C_Thread c3_thread = new C_Thread(c);
C_Thread c4_thread = new C_Thread(c);
C_Thread c5_thread = new C_Thread(c);
p_thread.start();
/* p2_thread.start();
p3_thread.start();
p4_thread.start();
p5_thread.start();*/
c_thread.start();
c2_thread.start();
c3_thread.start();
c4_thread.start();
c5_thread.start();
}
}

// 生产者
class P {
private ProductAndCustomer2 pc;

P(ProductAndCustomer2 pc) {
this.pc = pc;
}

public void pushService() {
pc.push();
}
}

// 生产者线程
class P_Thread extends Thread {
private P p;

P_Thread(P p) {
this.p = p;
}

@Override
public void run() {
super.run();
while (true) {
p.pushService();
}
}
}

// 消费者
class C {
private ProductAndCustomer2 pc;

C(ProductAndCustomer2 pc) {
this.pc = pc;
}

public void popService() {
System.out.println("pop--->" + pc.pop());
}
}

// 消费者线程
class C_Thread extends Thread {
private C c;

C_Thread(C c) {
this.c = c;
}

@Override
public void run() {
super.run();
while (true) {
c.popService();
}
}
}

（4）一生产与多消费：操作栈，解决wait条件改变 与  假死问题

        因为if条件发生改变时并没有得到及时的响应，所以多个呈wait状态的线程被唤醒，继而执行list.remove(0)代码而出现异常。 解决条件改变问题的方法为：将if (list.size() == 1) 改成 while(list.size() == 1)语句。

        解决假死问题：将两处this.notify(); 改写成 this.notifyAll();

（5）多生产与一消费：操作栈

（6）多生产与多消费：操作栈

10.  通过管道进行线程间通信

        管道流（pipeStream）是一种特殊的流，用于在不同线程间直接传送数据。一个线程发送（write）数据到输出管道，另一个线程从输入管道中读取（read）数据，实现不同线程之间的通信，而无需借助于类似临时文件之类的东西。

（1）通过管道进行线程间通信：字节流(byte)

        PipedInputStream和PipedOutputStream（两个线程通过管道流进行字节数据的传输）

(2)通过管道进行线程间通信：字符流（char)

        PipedReader和PipedWriter（两个线程通过管道流进行字节数据的传输）

阅读更多