Java并发编程之wait、notify和join原理

Java微观世界

发布于 2025-01-21 08:03:14

2630

文章被收录于专栏：springbootspringboot

1、wait、notify介绍

必须要获取到锁对象, 才能调用这些方法

当线程0获得到了锁, 成为Monitor的Owner, 但是此时它发现自己想要执行synchroized代码块的条件不满足; 此时它就调用obj.wait方法, 进入到Monitor中的WaitSet集合, 此时线程0的状态就变为WAITING
处于BLOCKED和WAITING状态的线程都为阻塞状态，CPU都不会分给他们时间片。但是有所区别： 1）BLOCKED状态的线程是在竞争锁对象时，发现Monitor的Owner已经是别的线程了，此时就会进入EntryList中，并处于BLOCKED状态 2）WAITING状态的线程是获得了对象的锁，但是自身的原因无法执行synchroized的临界区资源需要进入阻塞状态时，锁对象调用了wait方法而进入了WaitSet中，处于WAITING状态
处于BLOCKED状态的线程会在锁被释放的时候被唤醒
处于WAITING状态的线程只有被锁对象调用了notify方法(obj.notify/obj.notifyAll)，才会被唤醒。然后它会进入到EntryList, 重新竞争锁 (此时就将锁升级为重量级锁)

2、API介绍

下面的三个方法都是Object中的方法; 通过锁对象来调用

wait(): 让获得对象锁的线程到waitSet中一直等待
wait(long n) : 当该等待线程没有被notify, 等待时间到了之后, 也会自动唤醒
notify(): 让获得对象锁的线程, 使用锁对象调用notify去waitSet的等待线程中挑一个唤醒
notifyAll() : 让获得对象锁的线程, 使用锁对象调用notifyAll去唤醒waitSet中所有的等待线程

3、sleep(long n) 和 wait(long n)的区别

不同点

sleep是Thread类的静态方法，wait是Object的方法，Object又是所有类的父类，所以所有类都有wait方法
sleep在阻塞的时候不会释放锁，而wait在阻塞的时候会释放锁
sleep不需要强制和synchronized配合使用，但wait需要和synchronized一起用

同点

阻塞状态都为TIMED_WAITING (限时等待)

4、wait/notify的正确使用

为什么 if会出现虚假唤醒？

因为if只会执行一次，执行完会接着向下执行if(){}后边的逻辑；
而while不会，直到条件满足才会向下执行while(){}后边的逻辑

使用while循环去循环判断一个条件，而不是使用if只判断一次条件；即wait()要在while循环中

public class SpuriousWakeup {

    public static void main(String[] args) {
        WareHouse wareHouse = new WareHouse();
        Producer producer = new Producer(wareHouse);
        Customer customer = new Customer(wareHouse);

        new Thread(producer, "ProducerA").start();
        new Thread(producer, "ProducerB").start();

        new Thread(customer, "ConsumerC").start();
        new Thread(customer, "ConsumerD").start();
    }
}

// 仓库
class WareHouse {
    private volatile int product = 0;

    // 入库
    public synchronized void purchase() {
        // 库存已满，仓库最多容纳1个货品
        while (product > 0) {
            System.out.println(Thread.currentThread().getName() + ": " + "已满！");
            try {
                this.wait();
            } catch (InterruptedException e) {
                // ignore exception
            }
        }
        ++product;
        // 该线程从while中出来的时候，已满足条件
        System.out.println(Thread.currentThread().getName() + ": " + "-------------入库成功，余货：" + product);
        this.notifyAll();
    }

    // 出库
    public synchronized void outbound() {
        while (product <= 0) {
            System.out.println(Thread.currentThread().getName() + ": " + "库存不足，无法出库");
            try {
                this.wait();
            } catch (InterruptedException e) {
                // ignore exception
            }
        }
        --product;
        System.out.println(Thread.currentThread().getName() + ":出库成功，余货：" + product);
        this.notifyAll();
    }
}

// 生产者
class Producer implements Runnable {
    private WareHouse wareHouse;

    public Producer(WareHouse wareHouse) {
        this.wareHouse = wareHouse;
    }

    @Override
    public void run() {
        for (int i = 0; i < 5; ++i) {
            try {
                Thread.sleep(200);
            } catch (InterruptedException e) {
            }
            wareHouse.purchase();
        }
    }
}

// 消费者
class Customer implements Runnable {
    private WareHouse wareHouse;

    public Customer(WareHouse wareHouse) {
        this.wareHouse = wareHouse;
    }

    @Override
    public void run() {
        for (int i = 0; i < 5; ++i) {
            wareHouse.outbound();
        }
    }
}

输出结果：

# 2、join源码

示例

先打印t1，1s后打印mian

public class TestJoin {
    public static void main(String[] args) throws InterruptedException {

        Thread t1 = new Thread(() -> {
            System.out.println("t1");//先执行
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "t1");
        t1.start();

        t1.join();
        System.out.println("main");//一秒后执行
    }
}

join()

当millis为0时，如果线程存活则将调用native方法wait，在main线程中t1.join()，则main线程进入等待阻塞状态
t1执行完毕，或者t1被打断，则唤醒main线程

join(long)

delay：需要暂停时间-已经暂停时间 = 剩余需要暂停时间
wait(delay)等待期间被唤醒，则往下执行，now记录已经暂停时间
下次进入while则只暂停delay，而不用暂停millis秒

public final synchronized void join(long millis)
throws InterruptedException {
    long base = System.currentTimeMillis();//起始时间
    long now = 0;//已经暂停时间

    if (millis < 0) {
        throw new IllegalArgumentException("timeout value is negative");
    }

    if (millis == 0) {
        while (isAlive()) {
            wait(0);
        }
    } else {
        while (isAlive()) {
            long delay = millis - now;
            if (delay <= 0) {
                break;
            }
            wait(delay);
            now = System.currentTimeMillis() - base;
        }
    }
}

public final native void wait(long timeout) throws InterruptedException;

3、park & unpack

3.1、基本使用

park/unpark都是LockSupport类中的的方法
先调用unpark后,再调用park, 此时park不会暂停线程

// 暂停当前线程
LockSupport.park();
// 恢复某个线程的运行
LockSupport.unpark(thread);

3.2、 park、 unpark 原理

每个线程都有自己的一个 Parker 对象，由三部分组成 _counter， _cond和 _mutex

打个比喻线程就像一个旅人，Parker 就像他随身携带的背包，条件变量 _ cond就好比背包中的帐篷。_counter 就好比背包中的备用干粮（0 为耗尽，1 为充足）
调用 park 就是要看需不需要停下来歇息 1）如果备用干粮耗尽，那么钻进帐篷歇息 2）如果备用干粮充足，那么不需停留，继续前进
调用 unpark，就好比使干粮充足 1）如果这时线程还在帐篷，就唤醒让他继续前进 2）如果这时线程还在运行，那么下次他调用 park 时，仅是消耗掉备用干粮，不需停留继续前进 3）因为背包空间有限，多次调用 unpark 仅会补充一份备用干粮

先调用park再调用upark的过程

先调用park的情况

当前线程调用 Unsafe.park() 方法
检查 _counter, 本情况为0, 这时, 获得_mutex 互斥锁(mutex对象有个等待队列 _cond)
线程进入 _cond 条件变量阻塞
设置_counter = 0 (没干粮了)

调用unpark

调用Unsafe.unpark(Thread_0)方法，设置_counter 为 1
唤醒 _cond 条件变量中的 Thread_0
Thread_0 恢复运行
设置 _counter 为 0

先调用upark再调用park的过程

调用 Unsafe.unpark(Thread_0)方法，设置 _counter 为 1
当前线程调用 Unsafe.park() 方法
检查 _counter，本情况为 1，这时线程无需阻塞，继续运行
设置 _counter 为 0

4、线程状态转换

1、 NEW <–> RUNNABLE

t.start()方法时, NEW --> RUNNABLE

2、RUNNABLE <–> WAITING

线程用synchronized(obj)获取了对象锁后
调用 obj.wait()方法时，t 线程进入waitSet中, 从RUNNABLE --> WAITING
调用 obj.notify()，obj.notifyAll()，t.interrupt() 时, 唤醒的线程都到entrySet阻塞队列成为BLOCKED状态, 在阻塞队列,和其他线程再进行竞争锁 1）竞争锁成功，t 线程从 WAITING --> RUNNABLE 2）竞争锁失败，t 线程从 WAITING --> BLOCKED

3、RUNNABLE <–> WAITING

当前线程调用 t.join() 方法时，当前线程从 RUNNABLE --> WAITING ,注意是当前线程在t线程对象在waitSet上等待
t 线程运行结束，或调用了当前线程的 interrupt() 时，当前线程从 WAITING --> RUNNABLE

4、RUNNABLE <–> WAITING

当前线程调用 LockSupport.park() 方法会让当前线程从RUNNABLE --> WAITING
调用 LockSupport.unpark(目标线程) 或调用了线程的 interrupt() ，会让目标线程从 WAITING --> RUNNABLE

5、RUNNABLE <–> TIMED_WAITING (带超时时间的wait)

t 线程用synchronized(obj) 获取了对象锁后
调用 obj.wait(long n) 方法时，t 线程从 RUNNABLE --> TIMED_WAITING
t 线程等待时间超过了 n 毫秒，或调用 obj.notify() ， obj.notifyAll() ， t.interrupt() 时; 唤醒的线程都到entrySet阻塞队列成为BLOCKED状态, 在阻塞队列,和其他线程再进行竞争锁 1）竞争锁成功，t 线程从 TIMED_WAITING --> RUNNABLE 2）竞争锁失败，t 线程从 TIMED_WAITING --> BLOCKED

6、RUNNABLE <–> TIMED_WAITING

当前线程调用 t.join(long n) 方法时，当前线程从 RUNNABLE --> TIMED_WAITING 注意是当前线程在t 线程对象的waitSet等待
当前线程等待时间超过了 n 毫秒，或t 线程运行结束，或调用了当前线程的 interrupt() 时，当前线程从 TIMED_WAITING --> RUNNABLE

7、RUNNABLE <–> TIMED_WAITING

当前线程调用 Thread.sleep(long n) ，当前线程从 RUNNABLE --> TIMED_WAITING
当前线程等待时间超过了 n 毫秒或调用了线程的 interrupt() ，当前线程从 TIMED_WAITING --> RUNNABLE

8、RUNNABLE <–> TIMED_WAITING

当前线程调用 LockSupport.parkNanos(long nanos) 或 LockSupport.parkUntil(long millis) 时，当前线程从 RUNNABLE --> TIMED_WAITING
调用LockSupport.unpark(目标线程) 或调用了线程的 interrupt() ，或是等待超时，会让目标线程从 TIMED_WAITING–> RUNNABLE

9、RUNNABLE <–> BLOCKED

t 线程用 synchronized(obj) 获取了对象锁时如果竞争失败，从 RUNNABLE –> BLOCKED, 持 obj 锁线程的同步代码块执行完毕，会唤醒该对象上所有 BLOCKED 的线程重新竞争
如果其中 t 线程竞争成功，从 BLOCKED –> RUNNABLE ，其它失败的线程仍然 BLOCKED

10、 RUNNABLE <–> TERMINATED