Java多线程竞速：利用wait()和notify()获取首个结果并协调线程-java教程-PHP中文网

Java多线程竞速：利用wait()和notify()获取首个结果并协调线程

DDD

发布： 2025-11-11 23:22:01

原创

291人浏览过

java多线程竞速：利用wait()和notify()获取首个结果并协调线程

本文深入探讨了在Java多线程环境中，当多个线程竞相执行任务，且仅需获取最先完成任务的线程所产生的结果时，如何高效地进行线程协调。通过共享状态变量、`synchronized`关键字以及`wait()`和`notifyAll()`机制，文章详细阐述了如何设计工作线程和主线程的交互逻辑，以确保只采纳首个有效结果，并优化其他线程的执行，避免不必要的计算。

引言：多线程竞速与结果获取

在并发编程中，一种常见的场景是启动多个线程来执行相似或互补的任务，目标是尽快获得一个结果。一旦某个线程成功地计算出结果，其他仍在运行的线程就变得不再重要，我们希望能够及时停止它们或至少阻止它们继续不必要的计算。例如，一个线程从用户输入（Scanner）获取数据，另一个线程通过键盘监听器获取数据，我们只需要其中任何一个线程首先提供的数据。

这种场景的核心挑战在于：

线程间的通信：如何让主线程知道某个工作线程已经找到了解决方案？
竞态条件处理：如何确保即使多个线程几乎同时找到解决方案，也只有一个解决方案被采纳？
资源优化：如何让其他未获胜的线程停止工作，避免浪费CPU周期？

传统的Thread.currentThread.wait()用法通常是错误的，因为它会导致当前线程在自身对象上等待，而没有其他线程可以notify()它。正确的做法是使用一个共享的锁对象，并配合synchronized块来协调wait()和notify()调用。

立即学习“Java免费学习笔记（深入）”；

核心机制：共享状态、同步与通知

为了解决上述问题，我们需要引入以下核心机制：

协和·太初

国内首个针对罕见病领域的AI大模型

查看详情

共享状态变量：用于存储解决方案和指示解决方案是否已找到的标志。
互斥锁对象：一个独立的Object实例，作为synchronized块的锁，用于保护共享状态变量的访问，并作为wait()和notifyAll()调用的目标。

// 共享状态变量
volatile boolean solutionIsFound = false; // 使用volatile确保可见性
SolutionType solution; // 存储找到的解决方案
final Object solutionMutex = new Object(); // 用于同步的锁对象

登录后复制

volatile关键字在这里可以确保solutionIsFound的最新值对所有线程立即可见。虽然synchronized块本身也提供了内存可见性保证，但在此处明确使用volatile可以更清晰地表达其意图。

工作线程的设计与实现

每个工作线程（如从Scanner或KeyListener获取输入的线程）都需要遵循以下逻辑：

在工作开始前检查：进入工作循环后，首先检查是否已有其他线程找到了解决方案。如果solutionIsFound为true，则当前线程应立即退出，避免不必要的计算。
执行工作增量：进行一部分计算或尝试获取输入。
找到解决方案后更新状态：如果当前线程成功找到了解决方案，它需要：
- 获取solutionMutex锁。
- 再次检查solutionIsFound：这是“双重检查锁定”模式的一种变体，以防止在获取锁之前，另一个线程已经更新了状态。
- 如果确实是当前线程率先找到，则更新solutionIsFound为true，存储solution，并通过solutionMutex.notifyAll()唤醒所有等待在该锁上的线程（包括主线程）。
- 然后，当前线程退出。

以下是工作线程的示例代码结构：

class WorkerThread implements Runnable {
    // 共享变量，通过构造函数注入
    private volatile boolean sharedSolutionIsFound;
    private SolutionType sharedSolution;
    private final Object sharedSolutionMutex;

    // 构造函数用于初始化共享变量
    public WorkerThread(volatile boolean solutionIsFound, SolutionType solution, Object solutionMutex) {
        this.sharedSolutionIsFound = solutionIsFound;
        this.sharedSolution = solution;
        this.sharedSolutionMutex = solutionMutex;
    }

    @Override
    public void run() {
        while (true) {
            // 第一层检查：在执行耗时操作前，快速判断是否已找到解决方案
            synchronized (sharedSolutionMutex) {
                if (sharedSolutionIsFound) {
                    // 其他线程已获胜，当前线程无需继续
                    System.out.println(Thread.currentThread().getName() + " 检测到解决方案已找到，退出。");
                    return;
                }
            }

            // ... 执行一些工作增量 ...
            // 假设这里模拟获取输入或进行计算
            SolutionType currentThreadResult = simulateWork(); 

            if (currentThreadResult != null) { // 如果当前线程找到了解决方案
                synchronized (sharedSolutionMutex) {
                    if (!sharedSolutionIsFound) { // 再次检查，确保是第一个发现的
                        sharedSolutionIsFound = true;
                        sharedSolution = currentThreadResult;
                        System.out.println(Thread.currentThread().getName() + " 找到了解决方案: " + sharedSolution);
                        sharedSolutionMutex.notifyAll(); // 唤醒所有等待的线程（包括主线程）
                    } else {
                        System.out.println(Thread.currentThread().getName() + " 找到了解决方案，但其他线程更快。");
                    }
                }
                return; // 线程完成任务，退出
            }
        }
    }

    private SolutionType simulateWork() {
        try {
            // 模拟耗时操作，例如从Scanner读取或等待KeyListener事件
            Thread.sleep((long) (Math.random() * 2000)); // 随机延迟0-2秒
            if (Math.random() > 0.7) { // 模拟有30%的几率找到解决方案
                return new SolutionType("Solution from " + Thread.currentThread().getName());
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.out.println(Thread.currentThread().getName() + " 被中断。");
        }
        return null;
    }
}

// 示例 SolutionType 类，用于表示解决方案的数据类型
class SolutionType {
    String value;
    public SolutionType(String value) { this.value = value; }
    @Override public String toString() { return value; }
}

登录后复制