在单线程编程中，要执行的操作需要满足一定的条件才能执行，可以把这个操作放在if语句块中。
在多线程编程中，可能A线程的条件没有满足只是暂时的，稍后其他的线程B可能会更新条件使得A线程的条件得到满足，可以将A线程暂停，直到他的条件得到满足后再将A线程唤醒。他的伪代码：
atomics { //原子操作
while(条件不成立) {
等待
}
当前线程被唤醒条件满足后，继续执行下面的操作
}
2，等待/通知机制的实现
2.1，wait()
Object类中的wait()方法可以使执行当前代码的线程等待，暂停执行，知道接到通知或被中断为止。
注意：
- wait()方法只能在同步代码块中由锁对象调用
- 调用wait()方法，当前线程会释放锁
其伪代码如下：
//在调用wait()方法前获得对象的内部锁
synchronized(锁对象){
while(条件不成立){
//通过锁对象调用wait()方法暂停线程，会释放锁对象
锁对象.wait();
}
//线程的条件满足了继续向下执行
}
示例
public class Test02 {
public static void main(String[] args) {
try {
String text = "drhj";
String another = "hello";
System.out.println("同步前的代码");
synchronized (text) {
System.out.println("同步代码块开始....");
text.wait(); //调用wait()方法后，当前线程就会等待，释放锁对象，当前线程需要被唤醒，如果没有唤醒就会一直等待
System.out.println("wait后面的代码....");
}
System.out.println("同步代码块后面的代码");
} catch (Exception e) {
e.printStackTrace();
}
}
}
可见当前线程处于等待状态
Object类的notify()可以唤醒线程，该方法也必须在同步代码块中由锁对象调用。没有使用锁对象调用wait()/notify()会抛出IllegalMonitorStateException异常。如果有多个等待的线程，notify()方法只能唤醒其中的一个。在同步代码块中调用notify()方法后不会立即释放锁对象，需要等当前同步代码块执行完后才会释放锁对象，一般将notify()方法放在同步代码块的最后。他的伪代码如下：
synchronized(锁对象) {
//执行修改保护条件的代码
//唤醒其他线程
锁对象.notify();
}
示例
public class Test03 {
public static void main(String[] args) throws InterruptedException {
String lock = "drhj"; //定义一个字符串作为锁对象
Thread t1 = new Thread(()-> {
synchronized (lock) {
System.out.println("线程1开始等待：" + System.currentTimeMillis());
try {
lock.wait(); //等待线程，会释放锁对象，当前线程转入waiting状态
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("线程1结束等待：" + System.currentTimeMillis());
}
});
//唤醒第一个线程
Thread t2 = new Thread(()-> {
//notify()方法也需要在同步代码块中，由锁对象调用
synchronized (lock) {
System.out.println("线程2开始唤醒：" + System.currentTimeMillis());
lock.notify(); //唤醒在lock锁对象上等待的线程
System.out.println("线程2结束线程唤醒：" + System.currentTimeMillis());
}
});
t1.start();
Thread.sleep(3000); //main线程睡眠3s，确保t1入睡
t2.start(); //t1线程开启唤醒
}
}
可见，在线程1在等待时，只有被唤醒时才会继续执行。
notify不会立即释放锁对象
public class Test04 {
public static void main(String[] args) throws InterruptedException {
//定义一个List集合存储String数据
List list = new ArrayList<>();
//定义第一个线程，当list集合中元素的数量不等于5时线程等待
Thread t1 = new Thread(()->{
synchronized (list) {
if (list.size() != 5) {
System.out.println("线程1开始等待：" + System.currentTimeMillis());
try {
list.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("线程1被唤醒：" + System.currentTimeMillis());
}
}
});
//定义第二个线程,向list集合中添加元素
Thread t2 = new Thread(()->{
synchronized (list) {
for (int i = 0; i < 10; i++) {
list.add("data--" + i);
System.out.println("线程2添加了第" + (i + 1) + "个数据");
//判断元素的数量是否满足唤醒线程1
if (list.size() == 5) {
list.notify(); //唤醒线程,不会立即释放锁对象
System.out.println("线程2已经发现唤醒通知");
}
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
});
t1.start();
//为了确保t2在t1之后开启，即让t1线程先睡眠
Thread.sleep(500);
t2.start();
}
}
按理说在添加第5个数据时，线程1就被唤醒了，但实际上是在线程2执行完成后才唤醒线程1，说明notify不会唤醒锁对象。
当线程处于wait()等待状态时，调用线程对象的interrupt()方法会中断线程的等待状态，会产生InterruptedException异常
public class Test05 {
public static void main(String[] args) {
SubThread t = new SubThread();
t.start();
try {
Thread.sleep(2000); //主线程睡眠2s,确保子线程处于wait等待状态
t.interrupt();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
private static final Object LOCK = new Object(); //定义常量作为锁对象
static class SubThread extends Thread {
@Override
public void run() {
synchronized (LOCK) {
try {
System.out.println("begin wait...");
LOCK.wait();
System.out.println("end wait...");
} catch (Exception e) {
System.out.println("wait等待被中断了*****");
}
}
}
}
}
2.4，notify()与notifyAll()
notify()一次只能唤醒一个线程，如果有多个等待的线程，只能随机唤醒其中的某一个；想要唤醒所有等待线程，需要调用notifyAll()。
public class Test06 {
public static void main(String[] args) throws InterruptedException {
Object lock = new Object();
SubThread t1 = new SubThread(lock);
SubThread t2 = new SubThread(lock);
SubThread t3 = new SubThread(lock);
t1.setName("t1");
t2.setName("t2");
t3.setName("t3");
t1.start();
t2.start();
t3.start();
Thread.sleep(2000);
//调用notify()唤醒子线程
synchronized (lock) {
lock.notify();
}
}
static class SubThread extends Thread {
private Object lock; //定义实例变量作为锁对象
public SubThread(Object lock) {
this.lock = lock;
}
@Override
public void run() {
synchronized (lock) {
try {
System.out.println(Thread.currentThread().getName() + " -- begin wait...");
lock.wait();
System.out.println(Thread.currentThread().getName() + "--end wait...");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
notify只会唤醒其中一个，且是随机的。
public class Test06 {
public static void main(String[] args) throws InterruptedException {
Object lock = new Object();
SubThread t1 = new SubThread(lock);
SubThread t2 = new SubThread(lock);
SubThread t3 = new SubThread(lock);
t1.setName("t1");
t2.setName("t2");
t3.setName("t3");
t1.start();
t2.start();
t3.start();
Thread.sleep(2000);
//调用notify()唤醒子线程
synchronized (lock) {
lock.notifyAll();
}
}
static class SubThread extends Thread {
private Object lock; //定义实例变量作为锁对象
public SubThread(Object lock) {
this.lock = lock;
}
@Override
public void run() {
synchronized (lock) {
try {
System.out.println(Thread.currentThread().getName() + " -- begin wait...");
lock.wait();
System.out.println(Thread.currentThread().getName() + "--end wait...");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
notifyAll()唤醒所有的线程。
wait(long) 带有long类型参数的wait()等待，如果在参数指定的时间内没有被唤醒，超时后会自动唤醒
public class Test07 {
public static void main(String[] args) {
final Object obj = new Object();
Thread t = new Thread(()-> {
synchronized (obj) {
try {
System.out.println("thread begin wait");
obj.wait(5000); //如果5000ms内没有被唤醒，会自动唤醒
System.out.println("end wait...");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
});
}
}
过来5000ms后线程被唤醒
线程wait()等待后，可以调用notify()唤醒线程，如果notify()唤醒的过早，在等待之前就调用notify()可能会打乱程序正常的执行逻辑.
public class Test08 {
public static void main(String[] args) {
final Object Lock = new Object();
Thread t1 = new Thread(()-> {
synchronized (Lock) {
try {
System.out.println("begin wait");
Lock.wait();
System.out.println("wait end...");
} catch (Exception e) {
e.printStackTrace();
}
}
});
Thread t2 = new Thread(()-> {
synchronized (Lock) {
System.out.println("begin notify");
Lock.notify();
System.out.println("end notify");
}
});
//如果先开启t1，再开启t2线程，大多数情况下，t1先等待，t2再把t1唤醒
// t1.start();
// t2.start();
//如果先开启t2通知线程，再开启t1等待线程，可能会出现t1线程等待没有收到通知的情况
t2.start();
t1.start();
//实际上，调用start()就是告诉线程调度器，当前线程准备就绪，线程调度器在什么时候开启线程不确定，即调用start()方法的顺序，并不一定就是线程实际开启的顺序。
//在当前示例中，t1等待后让t2线程唤醒，如果t2线程先唤醒了，就不让t1线程等待了
}
}
解决方法
public class Test08 {
static boolean isFirst = true; //定义静态变量作为是否第一个运行的线程标志
public static void main(String[] args) {
final Object Lock = new Object();
Thread t1 = new Thread(()-> {
synchronized (Lock) {
while (isFirst) {
try {
System.out.println("begin wait");
Lock.wait();
System.out.println("wait end...");
} catch (Exception e) {
e.printStackTrace();
}
}
}
});
Thread t2 = new Thread(()-> {
synchronized (Lock) {
System.out.println("begin notify");
Lock.notify();
System.out.println("end notify");
isFirst = false; //通知后，就把第一个线程标志修改为false
}
});
//如果先开启t1，再开启t2线程，大多数情况下，t1先等待，t2再把t1唤醒
// t1.start();
// t2.start();
//如果先开启t2通知线程，再开启t1等待线程，可能会出现t1线程等待没有收到通知的情况
t2.start();
t1.start();
//实际上，调用start()就是告诉线程调度器，当前线程准备就绪，线程调度器在什么时候开启线程不确定，即调用start()方法的顺序，并不一定就是线程实际开启的顺序。
//在当前示例中，t1等待后让t2线程唤醒，如果t2线程先唤醒了，就不让t1线程等待了
}
}
2.7，wait等待条件反生了变化
在使用wait/notify模式时，注意wait条件发生了变化，也可能会造成逻辑的混乱。
测试一：先开启添加数据的线程，再开启一个取数据的线程，大多数情况下正常执行
public class Test09 {
public static void main(String[] args) {
ThreadAdd t1 = new ThreadAdd();
ThreadSubtract t2 = new ThreadSubtract();
t2.setName("subtract 1");
//测试一：先开启添加数据的线程，再开启一个取数据的线程，大多数情况下正常执行
t1.start();
t2.start();
}
static List list = new ArrayList<>();
//取数据
public static void subtract() {
synchronized (list) {
if (list.size() == 0) {
try {
System.out.println(Thread.currentThread().getName() + " begin wait...");
list.wait();
System.out.println(Thread.currentThread().getName() + " end wait...");
} catch (Exception e) {
e.printStackTrace();
}
}
Object data = list.remove(0); //从集合中取数据
System.out.println(Thread.currentThread().getName() + " 从集合中取了" + data + "后，集合中数据的数量：" + list.size());
}
}
//添加数据
public static void add() {
synchronized (list) {
list.add("data");
list.notifyAll();
}
}
//添加数据线程
static class ThreadAdd extends Thread {
@Override
public void run() {
add();
}
}
//取数据线程
static class ThreadSubtract extends Thread {
@Override
public void run() {
subtract();
}
}
}
因为大概率会先添加元素，所以取数据的时候不会发生等待。
测试二：先开启取数据的线程，再开启添加数据的线程，取数据的线程会先等待，等到添加数据之后，再取数据
public class Test09 {
public static void main(String[] args) {
ThreadAdd t1 = new ThreadAdd();
ThreadSubtract t2 = new ThreadSubtract();
t2.setName("subtract 1");
//测试二：先开启取数据的线程，再开启添加数据的线程，取数据的线程会先等待，等到添加数据之后，再取数据
t2.start();
t1.start();
}
static List list = new ArrayList<>();
//取数据
public static void subtract() {
synchronized (list) {
if (list.size() == 0) {
try {
System.out.println(Thread.currentThread().getName() + " begin wait...");
list.wait();
System.out.println(Thread.currentThread().getName() + " end wait...");
} catch (Exception e) {
e.printStackTrace();
}
}
Object data = list.remove(0); //从集合中取数据
System.out.println(Thread.currentThread().getName() + " 从集合中取了" + data + "后，集合中数据的数量：" + list.size());
}
}
//添加数据
public static void add() {
synchronized (list) {
list.add("data");
list.notifyAll();
}
}
//添加数据线程
static class ThreadAdd extends Thread {
@Override
public void run() {
add();
}
}
//取数据线程
static class ThreadSubtract extends Thread {
@Override
public void run() {
subtract();
}
}
}
先执行取数据的时候，会发生等待状态。
测试三：开启两个取数据的线程，再开启添加数据的线程
public class Test09 {
public static void main(String[] args) {
ThreadAdd t1 = new ThreadAdd();
ThreadSubtract t2 = new ThreadSubtract();
t2.setName("subtract 1");
//测试三：开启两个取数据的线程，再开启添加数据的线程
ThreadSubtract t3 = new ThreadSubtract();
t3.setName("subtract 2");
t2.start();
t3.start();
t1.start();
}
static List list = new ArrayList<>();
//取数据
public static void subtract() {
synchronized (list) {
if (list.size() == 0) {
try {
System.out.println(Thread.currentThread().getName() + " begin wait...");
list.wait();
System.out.println(Thread.currentThread().getName() + " end wait...");
} catch (Exception e) {
e.printStackTrace();
}
}
Object data = list.remove(0); //从集合中取数据
System.out.println(Thread.currentThread().getName() + " 从集合中取了" + data + "后，集合中数据的数量：" + list.size());
}
}
//添加数据
public static void add() {
synchronized (list) {
list.add("data");
list.notifyAll();
}
}
//添加数据线程
static class ThreadAdd extends Thread {
@Override
public void run() {
add();
}
}
//取数据线程
static class ThreadSubtract extends Thread {
@Override
public void run() {
subtract();
}
}
}
可能执行顺序：
t2线程先启动，取数据时，集合中没有数据，wait()等待；t1线程获得CPU执行权，添加数据，把t2线程唤醒；t3线程开启后获得CPU执行权，正常取数据；t2线程获得CPU执行权，打印end wait…，然后再打印list.remove(0)取数据时，现在list集合中已经没有数据了，这时会产生异常。
出现异常的原因是：向list集合中添加了一个数据，remove()了两次。
解决办法：当等待线程被唤醒后，再判断一次集合中是否有数据可取。
//取数据
public static void subtract() {
synchronized (list) {
//if (list.size() == 0) {
while (list.size() == 0) {
try {
System.out.println(Thread.currentThread().getName() + " begin wait...");
list.wait();
System.out.println(Thread.currentThread().getName() + " end wait...");
} catch (Exception e) {
e.printStackTrace();
}
}
Object data = list.remove(0); //从集合中取数据
System.out.println(Thread.currentThread().getName() + " 从集合中取了" + data + "后，集合中数据的数量：" + list.size());
}
}
2.8，生产者消费者模式
在java中，负责产生数据的模块是生产者，负责使用数据的模块是消费者。生产者消费者解决数据的平衡问题，即先有数据然后才能使用，没有数据时，消费者需要等待。
- 生产-消费：操作值
public class ValueOP {
private String value = "";
//定义方法修改value的值
public void setValue() {
//如果value值不是""空串就等待
synchronized (this) {
if (!value.equalsIgnoreCase("")) {
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//设置value的值
String value= System.currentTimeMillis() + " - " + System.nanoTime();
System.out.println("set设置的值是：" + value);
this.value = value;
this.notify();
}
}
//定义方法读取字段值
public void getValue() {
synchronized (this) {
//如果是空串就等待
if (value.equalsIgnoreCase("")) {
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//不是空串，读取，字段值
System.out.println("get的值是：" + value);
this.value = "";
this.notify();
}
}
}
- 生产者
public class ProducerThread extends Thread {
//生产者生产数据就是调用ValueOP类的setValue方法给value字段赋值
private ValueOP obj;
public ProducerThread(ValueOP obj) {
this.obj = obj;
}
@Override
public void run() {
while (true) {
obj.setValue();
}
}
}
- 消费者
public class ConsumerThread extends Thread {
//消费者使用数据，就是使用ValueOP类的value字段的值
private ValueOP obj;
public ConsumerThread(ValueOP obj) {
this.obj = obj;
}
@Override
public void run() {
while (true) {
obj.getValue();
}
}
}
- 一生产者一消费者
public class Test {
public static void main(String[] args) {
ValueOP vp = new ValueOP();
ProducerThread p = new ProducerThread(vp);
ConsumerThread c = new ConsumerThread(vp);
p.start();
c.start();
}
}
这里是一发通知，一接受通知
5. 多生产者多消费者
public class Test {
public static void main(String[] args) {
ValueOP vp = new ValueOP();
ProducerThread p1 = new ProducerThread(vp);
ProducerThread p2 = new ProducerThread(vp);
ProducerThread p3 = new ProducerThread(vp);
ConsumerThread c1 = new ConsumerThread(vp);
ConsumerThread c2 = new ConsumerThread(vp);
ConsumerThread c3 = new ConsumerThread(vp);
p1.start();
p2.start();
p3.start();
c1.start();
c2.start();
c3.start();
}
}
出现大量get值是空串的情况是因为一个消费者线程在唤醒后的瞬间，别的消费者线程执行了getValue()方法，导致value值为空串。
6. 将 if 改为 while - 发生假死
public class ValueOP {
private String value = "";
//定义方法修改value的值
public void setValue() {
//如果value值不是""空串就等待
synchronized (this) {
while (!value.equalsIgnoreCase("")) {
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//设置value的值
String value= System.currentTimeMillis() + " - " + System.nanoTime();
System.out.println("set设置的值是：" + value);
this.value = value;
this.notify();
}
}
//定义方法读取字段值
public void getValue() {
synchronized (this) {
//如果是空串就等待
while (value.equalsIgnoreCase("")) {
try {
this.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
//不是空串，读取，字段值
System.out.println("get的值是：" + value);
this.value = "";
this.notify();
}
}
}
发生原因，举例说明：当生产者1获取执行权，value获得值，然后生产者2获得执行权，生产者2等待，生产者3获得执行权，等待；消费者1获得执行权将value置为空串，然后唤醒的却是消费者2，消费者2等待，消费者3获得执行权，等待；当消费者1再次获得执行权，发生了等待，最后生产者1获得执行权，给了value的值，却唤醒了生产者2，生产者2等待，最后执行权再次来到生产者1时，也发生了等待，至此程序假死，故无法继续输出。
解决方法：将notify()改为notifyAll().
在java.io包中PipeStream管道流用于在线程之间传递数据。一个线程发送数据到输出管道，另外一个线程从输入管道中读取数据。相关的类包括：PipedInputStream和PipedOutputStream，PipedReader和PipedWriter。
public class Test {
public static void main(String[] args) {
//定义管道字节流
PipedInputStream in = new PipedInputStream();
PipedOutputStream out = new PipedOutputStream();
//在输入管道流与输出管道流之间建立连接
try {
in.connect(out);
} catch (IOException e) {
e.printStackTrace();
}
//创建线程向管道流中写入数据
new Thread(()-> writeData(out)).start();
new Thread(()-> readData(in)).start();
}
//定义方法向管道流中写入数据
public static void writeData(PipedOutputStream out) {
try {
//分别把0~100之间的数写入管道中
for (int i = 0; i < 100; i++) {
String data = "" + i;
out.write(data.getBytes()); //把字节数组写入到输出管道流中
}
out.close(); //关闭管道流
} catch (Exception e) {
e.printStackTrace();
}
}
//定义方法从管道流中读取数据
public static void readData(PipedInputStream in) {
byte[] bytes = new byte[1024];
try {
//从管道输入字节流中读取字节保存到字节数组中
int len = in.read(bytes); //返回读到的字节数，如果没有读到任何数据返回-1
while (len != -1) {
//把byte数组从0开始讲到len个字节转换为字符串打印
System.out.println(new String(bytes, 0, len));
len = in.read(bytes); //继续从管道中读取数据
}
in.close();
} catch (Exception e) {
e.printStackTrace();
}
}
}
