Java线程通信的精髓:解析通知等待机制的工作原理

本文涉及的产品
公共DNS(含HTTPDNS解析),每月1000万次HTTP解析
云解析 DNS,旗舰版 1个月
全局流量管理 GTM,标准版 1个月
简介: Java线程通信的精髓:解析通知等待机制的工作原理

通知/等待机制

存在这样一个场景,一个线程修改了一个对象的值,而另一个线程需要感知到变化后去做一些处理。这是一种典型的生产者和消费者模式,这种模式在功能层面可以实现解耦,体系结构上也具备良好的申缩性。

如何用多线程去实现这种呢?最简单的办法是让消费者线程不断地循环检查是否符合执行条件,例如下面的代码:

while (value != desire) {
    Thread.sleep(1000);
}
doSomething();

这种是非常 low 的写法,存在以下问题:

  1. 难以确保及时性:睡眠的时候基本不消化处理器资源,但是如果睡得太久,就不能及时发现变化。
  2. 难以降低开销:如果将睡眠时间降低,那么消费者可以更加迅速的感应到变化,但是却要消耗更多的处理器资源,造成浪费。

以上问题用这种方式似乎难以调和,不过 Java 通过内置的通知/等待机制就可以很好的解决这个问题。

等待和通知的相关方法

这些方法是任意 Java 对象都具备的,因为是被定义在最底层的 Object 方法上。

方法名称 描述
wait() 调用该方法后,线程进入 WAITING 状态,只有等待别的线程通知或被中断后才能返回,调用此方法后,会释放当前线程持有的对象锁。
wait(long) 超时等待一段时间,单位毫秒,如果过了这个时间还没有被通知,那么则会超时返回。
wait(long, int) 也是超时等待,不过可以自定义单位,最小可以达到纳秒。
notify() 通知一个在对象对象上等待的线程,使其从 wait 方法返回,而返回的前提是该线程获取到了对象的锁。
notifyAll() 通知所有等待在该对象的线程。

通知等待机制,是指一个线程 A 调用了对象 O 的 wait 方法后进入等待状态(相当于线程 A 在执行 wait 这行代码后卡了),而另一个线程 B 调用了对象 O 的 notify 或 notifyAll 后,线程 A 可以收到通知,从对象 O 的 wait 方法返回(从刚才卡住的位置继续执行),进而执行后续的操作。调用对象 O 的 wait、notify 方法就像是开关信号一样,用来完成等待方和通知方之间的交互工作。

实战案例

public class WaitNotifyDemo1 {
    public static Integer value;
    public static final Object lock = new Object();
    public static void main(String[] args) throws InterruptedException {
        new Thread(() -> {
            synchronized (lock) {
                if (value == null) {
                    try {
                        System.out.println("value 为空,开始等待...");
                        lock.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                System.out.println("接收到通知,线程继续执行,value = " + value);
            }
        }, "WaitThread").start();
        TimeUnit.SECONDS.sleep(1);
        new Thread(() -> {
            synchronized (lock) {
                System.out.println(Thread.currentThread().getName() + " 获取到锁");
                value = 10086;
                System.out.println("value 赋值完成!通知等待线程处理...");
                lock.notify();
                try {
                    TimeUnit.SECONDS.sleep(3);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                System.out.println(Thread.currentThread().getName() + " 释放了锁");
            }
        }, "NotifyThread").start();
    }
}

上述的代码需要注意的细节:

  1. 使用 wait、notify、notifyAll 时需要先对调用的对象加锁。
  2. 调用 wait 方法后,WaitThread 的状态由 RUNNING 转为 WAITING,并将当前的线程放在等待队列。
  3. NotifyThread 调用 notify 后,等待的线程依旧不会从 wait 方法返回,而是需要 NotifyThread 释放锁之后,等待的线程才有机会从 wait 放回(还要去竞争锁,不一定抢到)。
  4. notify 方法是将一个线程从等待队列移到同步队列,线程状态由 WAITING 变为 BLOCKED。而 notifyAll 是将等待队列中的所有线程移到同步队列,线程状态由 WAITING 变为 BLOCKED。
  5. 从 wait 方法返回的前提是获取到了对象的锁。
  6. wait、notify、notifyAll 必须依赖于 synchronized,即必须在同步方法/代码块中调用。

等待通知机制依托于同步机制 synchronized ,其目的是确保等待线程从 wait 方法返回时能够感知到其他线程对变量做出的修改。

使用可重入锁也可以实现上面的效果,代码如下:

public class WaitNotifyDemo2 {
    public static Integer value;
    public static final Lock lock = new ReentrantLock();
    public static final Condition condition = lock.newCondition();
    public static void main(String[] args) throws InterruptedException {
        new Thread(() -> {
            lock.lock();
            try {
                if (value == null) {
                    try {
                        System.out.println("value 为空,开始等待...");
                        condition.await();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                System.out.println("接收到通知,线程继续执行,value = " + value);
            } finally {
                lock.unlock();
            }
        }, "WaitThread").start();
        TimeUnit.SECONDS.sleep(1);
        new Thread(() -> {
            lock.lock();
            try {
                System.out.println(Thread.currentThread().getName() + " 获取到锁");
                value = 10086;
                System.out.println("value 赋值完成!通知等待线程处理...");
                condition.signal();
                try {
                    TimeUnit.SECONDS.sleep(3);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                System.out.println(Thread.currentThread().getName() + " 释放了锁");
            } finally {
                lock.unlock();
            }
        }, "NotifyThread").start();
    }
}

为什么 wait、notify 必须搭配 synchronized

首先,要明白,每个对象都可以被认为是一个监视器 Monitor,这个监视器由三部分组成(一个独占锁,一个同步队列,一个等待队列)。

注意是一个对象只能有一个独占锁,但是任意线程线程都可以拥有这个独占锁。

  1. 对于对象的非同步方法而言,任意时刻可以有任意个线程调用该方法,即普通方法同一时刻可以有多个线程调用。
  2. 对于对象的同步方法而言,只有拥有这个对象的独占锁才能调用这个同步方法。如果这个独占锁被其他线程占用,那么另外一个调用该同步方法的线程就会处于阻塞状态,此线程进入同步队列。

若一个拥有该独占锁的线程调用该对象同步方法的 wait 方法,则该线程会释放独占锁,并加入对象的等待队列。

某个线程调用 notify、notifyAll 方法是将等待队列的线程转移到同步队列,然后让他们竞争锁,所以这个调用线程本身必须拥有锁。

通知等待的范式

等待方

等待方遵循以下原则:

  1. 获取对象锁。
  2. 如果条件不满足,则调用对象锁的 wait 方法,被通知后仍要检查条件。
  3. 条件满足则执行对应的逻辑。
synchronized(对象) {
    while(条件不满足) {
        对象.wait();
    }
    条件满足对应的处理逻辑
}

通知方

通知方遵循以下原则:

  1. 获取对象锁。
  2. 改变条件。
  3. 通知一个/所有等待在此对象锁上的线程。
synchronized(对象) {
  改变条件
    对象.notify(); // 对象.notifyAll();
}

Thread#join

如果线程 A 执行了线程 ThreadB.join(); 语句,其含义是:当线程 A 等待 ThreadB 执行完毕后才从 ThreadB.join() 返回继续执行。

线程 Thread 除了提供 join 方法之外, 还提供了 join(long)、join(long, int) 两个具备超时属性的方法。表示如果超过了指定的时间,对应的线程还是没有终止,则会直接从超市方法中返回。

下面的程序将会利用 join 的特性,控制三个线程按照顺序执行依次输出 Main、A、B、C。

public class JoinDemo {
    public static void main(String[] args) {
        Thread main = Thread.currentThread();
        Thread a = new Thread(() -> {
            try {
                main.join();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println("A");
        }, "A");
        Thread b = new Thread(() -> {
            try {
                a.join();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println("B");
        }, "B");
        Thread c = new Thread(() -> {
            try {
                b.join();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println("C");
        }, "C");
        a.start();
        b.start();
        c.start();
        System.out.println("Main");
    }
}

每个线程都有自己的前驱线程:Main -> A -> B -> C,当前线程终止的前提是前驱线程终止,等待前驱线程终止之后才能从 join 语句返回并继续执行自身逻辑,这里面也是涉及到了通知等待机制,查看 Thread join 方法的源码即可看到:

// 加锁当前线程对象
public final synchronized void join() throws InterruptedException {
    // 条件不满足,继续等待
    while (isAlive()) {
        wait(0);
    }
    // 条件符合,方法返回
}

当线程终止的时候,会隐式调用线程自身的 notifyAll 方法,会通知所有等待在该线程上的线程,可以看到 join 的实现和刚刚我们总结的范式是一致的,即加锁、循环、处理逻辑。

具体是怎么隐式调用的?Java 代码中并没有找到对应的位置,其实这个逻辑在固化在了 JDK 底层,主要源码都在 src/hotspot/share/runtime/thread.cpp 文件中:

void Thread::call_run() {
  ...
  this->pre_run();
  this->run();
  this->post_run();
  ...
}
void JavaThread::post_run() {
  // 线程将要退出
  this->exit(false);
  ...
}
void JavaThread::exit(bool destroy_vm, ExitType exit_type) {
  ...
  // 线程将要退出
  ensure_join(this);
  ... 
}
static void ensure_join(JavaThread* thread) {
  ...
  // 调用自身的 notify_all 方法
  lock.notify_all(thread);
  ...
}

保证多个线程的执行顺序

刚刚我们介绍的 join 就可以控制多个线程的执行顺序,再回顾一下代码:

public class JoinDemo {
    public static void main(String[] args) {
        Thread main = Thread.currentThread();
        Thread a = new Thread(() -> {
            try {
                main.join();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println("A");
        }, "A");
        Thread b = new Thread(() -> {
            try {
                a.join();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            System.out.println("B");
        }, "B");
        a.start();
        b.start();
        System.out.println("Main");
    }
}

join 的本质其实还是内置的通知等待机制,所以我们使用原生的方式也是可以实现的。

public class JoinDemo2 {
    public static void main(String[] args) {
        Thread main = Thread.currentThread();
        Thread a = new Thread(() -> {
            synchronized (main) {
                while (main.getState() != Thread.State.TERMINATED) {
                    try {
                        main.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
            System.out.println("A");
        }, "A");
        Thread b = new Thread(() -> {
            synchronized (a) {
                while (a.getState() != Thread.State.TERMINATED) {
                    try {
                        a.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
            System.out.println("B");
        }, "B");
        a.start();
        b.start();
        System.out.println("Main");
    }
}

每个线程执行完后会默认调用自身线程的 notifyAll 方法,所以我这个代码就没有明确调用了。

相关文章
|
2天前
|
XML 安全 Java
Java反射机制:解锁代码的无限可能
Java 反射(Reflection)是Java 的特征之一,它允许程序在运行时动态地访问和操作类的信息,包括类的属性、方法和构造函数。 反射机制能够使程序具备更大的灵活性和扩展性
15 5
Java反射机制:解锁代码的无限可能
|
5天前
|
Java 调度
[Java]线程生命周期与线程通信
本文详细探讨了线程生命周期与线程通信。文章首先分析了线程的五个基本状态及其转换过程,结合JDK1.8版本的特点进行了深入讲解。接着,通过多个实例介绍了线程通信的几种实现方式,包括使用`volatile`关键字、`Object`类的`wait()`和`notify()`方法、`CountDownLatch`、`ReentrantLock`结合`Condition`以及`LockSupport`等工具。全文旨在帮助读者理解线程管理的核心概念和技术细节。
18 1
[Java]线程生命周期与线程通信
|
1天前
|
存储 缓存 安全
🌟Java零基础:深入解析Java序列化机制
【10月更文挑战第20天】本文收录于「滚雪球学Java」专栏,专业攻坚指数级提升,希望能够助你一臂之力,帮你早日登顶实现财富自由🚀;同时,欢迎大家关注&&收藏&&订阅!持续更新中,up!up!up!!
9 3
|
1天前
|
安全 Java UED
深入理解Java中的异常处理机制
【10月更文挑战第25天】在编程世界中,错误和意外是不可避免的。Java作为一种广泛使用的编程语言,其异常处理机制是确保程序健壮性和可靠性的关键。本文通过浅显易懂的语言和实际示例,引导读者了解Java异常处理的基本概念、分类以及如何有效地使用try-catch-finally语句来处理异常情况。我们将从一个简单的例子开始,逐步深入到异常处理的最佳实践,旨在帮助初学者和有经验的开发者更好地掌握这一重要技能。
7 2
|
3天前
|
Java 数据库连接 开发者
Java中的异常处理机制####
本文深入探讨了Java语言中异常处理的核心概念,通过实例解析了try-catch语句的工作原理,并讨论了finally块和throws关键字的使用场景。我们将了解如何在Java程序中有效地管理错误,提高代码的健壮性和可维护性。 ####
|
5天前
|
安全 Java 程序员
深入浅出Java中的异常处理机制
【10月更文挑战第20天】本文将带你一探Java的异常处理世界,通过浅显易懂的语言和生动的比喻,让你在轻松阅读中掌握Java异常处理的核心概念。我们将一起学习如何优雅地处理代码中不可预见的错误,确保程序的健壮性和稳定性。准备好了吗?让我们一起踏上这段旅程吧!
20 6
|
3天前
|
存储 运维 Java
💻Java零基础:深入了解Java内存机制
【10月更文挑战第18天】本文收录于「滚雪球学Java」专栏,专业攻坚指数级提升,希望能够助你一臂之力,帮你早日登顶实现财富自由🚀;同时,欢迎大家关注&&收藏&&订阅!持续更新中,up!up!up!!
15 1
|
6天前
|
存储 Java 程序员
Java面试加分点!一文读懂HashMap底层实现与扩容机制
本文详细解析了Java中经典的HashMap数据结构,包括其底层实现、扩容机制、put和查找过程、哈希函数以及JDK 1.7与1.8的差异。通过数组、链表和红黑树的组合,HashMap实现了高效的键值对存储与检索。文章还介绍了HashMap在不同版本中的优化,帮助读者更好地理解和应用这一重要工具。
20 5
|
5天前
|
Java 开发者 UED
Java中的异常处理机制及其重要性
【10月更文挑战第20天】 在Java编程中,异常处理是确保程序健壮性的关键。本文将探讨Java中的异常处理机制,包括其定义、类型、抛出和捕获异常的方法,以及如何自定义异常。通过实例说明,我们将展示异常处理在实际编程中的应用,帮助读者理解其在提高代码质量和稳定性方面的重要性。
10 0
|
22天前
|
存储 消息中间件 资源调度
C++ 多线程之初识多线程
这篇文章介绍了C++多线程的基本概念,包括进程和线程的定义、并发的实现方式,以及如何在C++中创建和管理线程,包括使用`std::thread`库、线程的join和detach方法,并通过示例代码展示了如何创建和使用多线程。
36 1
C++ 多线程之初识多线程

推荐镜像

更多