JAVA 多线程执行顺序不确定？使用 Phaser 协调阶段性任务

好的，我们开始今天的讲座，主题是： JAVA 多线程执行顺序不确定？使用 Phaser 协调阶段性任务。

在并发编程的世界里，多线程的引入是为了提高程序的执行效率，充分利用多核处理器的能力。然而，多线程的并发执行也带来了一个核心问题：线程执行顺序的不确定性。这种不确定性在某些场景下是允许的，但在另一些场景下，我们需要精确地控制线程的执行顺序，确保按照预定的阶段性步骤完成任务。Phaser 类就是 Java 并发包 java.util.concurrent 中专门用于协调这些阶段性任务的工具。

一、多线程执行顺序不确定的本质

Java 线程的执行由 JVM 的线程调度器控制。线程调度器根据线程的优先级、等待时间、系统资源等因素决定哪个线程获得 CPU 时间片。由于这些因素在运行时是动态变化的，因此线程的执行顺序是不确定的。

例如，考虑以下简单的多线程程序：

public class SimpleThreadExample {
    public static void main(String[] args) {
        for (int i = 0; i < 3; i++) {
            final int threadId = i;
            new Thread(() -> {
                System.out.println("Thread " + threadId + " is running");
            }).start();
        }
    }
}

这段代码创建了三个线程，每个线程打印一条消息。每次运行这段程序，输出的顺序都可能不同。这是因为每个线程获得 CPU 时间片的机会是随机的。

二、为什么需要协调阶段性任务？

在许多应用场景中，我们需要将一个复杂的任务分解为多个阶段，并且要求所有线程在完成一个阶段后才能进入下一个阶段。

例如：

• 游戏开发： 一款多人在线游戏，所有玩家需要先加载地图资源，才能开始游戏。
• 大数据处理： 数据处理流程通常包含数据读取、数据清洗、数据转换、数据写入等阶段。只有所有线程完成数据读取后，才能开始数据清洗。
• 科学计算： 复杂的科学计算通常需要将数据分成多个块，由多个线程并行计算。只有所有线程完成当前块的计算，才能开始下一块的计算。

如果没有有效的机制来协调这些阶段性任务，可能会导致以下问题：

• 数据不一致： 某个线程可能在其他线程完成数据准备之前就开始处理数据，导致数据不一致。
• 资源竞争： 多个线程可能同时访问共享资源，导致资源竞争和死锁。
• 程序逻辑错误： 程序无法按照预期的阶段性步骤执行，导致逻辑错误。

三、Phaser 类的基本概念和使用

Phaser 类提供了一种灵活的机制来协调多个线程的阶段性任务。它的核心思想是将任务划分为多个阶段（phase），并允许线程动态地注册（register）、注销（unregister）和等待（arriveAndAwaitAdvance）某个阶段的完成。

3.1 Phaser 的主要方法：

方法名	描述
register()	将当前线程注册到 Phaser，增加参与者（participant）的数量。
bulkRegister(int parties)	一次性注册多个参与者。
arrive()	线程到达当前阶段，但不等待其他线程。返回 Phaser 的当前阶段数。
arriveAndAwaitAdvance()	线程到达当前阶段，并等待其他线程到达。当所有注册的线程都到达后，Phaser 进入下一个阶段。该方法会阻塞线程，直到 Phaser 进入下一个阶段。返回 Phaser 进入的下一个阶段的阶段数。
arriveAndDeregister()	线程到达当前阶段，并从 Phaser 中注销。这会减少参与者的数量。
forceTermination()	强制终止 Phaser，将 Phaser 的状态设置为终止状态。所有等待的线程都会被唤醒，并抛出 IllegalStateException 异常。
getPhase()	返回 Phaser 的当前阶段数。
getRegisteredParties()	返回已注册的参与者数量。
getArrivedParties()	返回已到达当前阶段的参与者数量。
getUnarrivedParties()	返回尚未到达当前阶段的参与者数量。
isTerminated()	检查 Phaser 是否已终止。

3.2 基本使用示例：

import java.util.concurrent.Phaser;

public class PhaserExample {
    public static void main(String[] args) {
        Phaser phaser = new Phaser(3); // 注册 3 个参与者

        for (int i = 0; i < 3; i++) {
            final int threadId = i;
            new Thread(() -> {
                System.out.println("Thread " + threadId + " started");
                phaser.arriveAndAwaitAdvance(); // 等待所有线程到达第一阶段
                System.out.println("Thread " + threadId + " finished phase 1");
                phaser.arriveAndAwaitAdvance(); // 等待所有线程到达第二阶段
                System.out.println("Thread " + threadId + " finished phase 2");
                phaser.arriveAndDeregister(); // 注销线程
            }).start();
        }
    }
}

在这个例子中，我们创建了一个 Phaser 对象，并注册了 3 个参与者。每个线程都执行两个阶段的任务，并在每个阶段结束后调用 arriveAndAwaitAdvance() 方法等待其他线程。当所有线程都到达当前阶段后，arriveAndAwaitAdvance() 方法才会返回，线程才能继续执行。在线程完成所有阶段的任务后，调用 arriveAndDeregister() 方法注销线程。

四、Phaser 的高级用法

4.1 动态注册和注销线程

Phaser 的一个重要特性是它允许线程动态地注册和注销。这意味着我们可以在程序运行过程中动态地调整参与者的数量。

例如，我们可以根据任务的需要，在不同的阶段注册或注销线程。

import java.util.concurrent.Phaser;

public class PhaserDynamicExample {
    public static void main(String[] args) throws InterruptedException {
        Phaser phaser = new Phaser(1); // 初始注册 1 个参与者 (main 线程)
        System.out.println("Phaser started, phase: " + phaser.getPhase() + ", registered parties: " + phaser.getRegisteredParties());

        // 启动 2 个新的线程
        new Thread(() -> {
            phaser.register(); // 动态注册
            System.out.println("Thread 1 registered, phase: " + phaser.getPhase() + ", registered parties: " + phaser.getRegisteredParties());
            phaser.arriveAndAwaitAdvance();
            System.out.println("Thread 1 finished phase 1, phase: " + phaser.getPhase());
            phaser.arriveAndDeregister(); // 注销
            System.out.println("Thread 1 deregistered, phase: " + phaser.getPhase() + ", registered parties: " + phaser.getRegisteredParties());
        }).start();

        new Thread(() -> {
            phaser.register(); // 动态注册
            System.out.println("Thread 2 registered, phase: " + phaser.getPhase() + ", registered parties: " + phaser.getRegisteredParties());
            phaser.arriveAndAwaitAdvance();
            System.out.println("Thread 2 finished phase 1, phase: " + phaser.getPhase());
            phaser.arriveAndDeregister(); // 注销
            System.out.println("Thread 2 deregistered, phase: " + phaser.getPhase() + ", registered parties: " + phaser.getRegisteredParties());
        }).start();

        Thread.sleep(100); // 确保新线程启动并注册

        System.out.println("Main thread arriving, phase: " + phaser.getPhase() + ", registered parties: " + phaser.getRegisteredParties());
        phaser.arriveAndAwaitAdvance(); // 等待所有线程到达第一阶段
        System.out.println("Main thread finished phase 1, phase: " + phaser.getPhase());

        phaser.arriveAndDeregister(); //main 线程也注销
        System.out.println("Main thread deregistered, phase: " + phaser.getPhase() + ", registered parties: " + phaser.getRegisteredParties());
    }
}

在这个例子中，main 线程首先注册自己，然后启动两个新的线程。这两个线程在启动后动态地注册到 Phaser。当所有线程都到达第一阶段后，main 线程才能继续执行。

4.2 Phaser 的 onAdvance() 方法

Phaser 类提供了一个 onAdvance() 方法，允许我们在每个阶段结束后执行一些自定义的操作。onAdvance() 方法的签名如下：

protected boolean onAdvance(int phase, int registeredParties)

phase 参数表示当前阶段数，registeredParties 参数表示注册的参与者数量。如果 onAdvance() 方法返回 true，则 Phaser 将被终止。

例如，我们可以使用 onAdvance() 方法来打印每个阶段的信息，并在所有线程都完成后终止 Phaser。

import java.util.concurrent.Phaser;

public class PhaserOnAdvanceExample {
    public static void main(String[] args) {
        Phaser phaser = new Phaser(3) {
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                System.out.println("Phase " + phase + " completed, registered parties: " + registeredParties);
                return registeredParties == 0; // 如果没有注册的参与者，则终止 Phaser
            }
        };

        for (int i = 0; i < 3; i++) {
            final int threadId = i;
            new Thread(() -> {
                System.out.println("Thread " + threadId + " started");
                phaser.arriveAndAwaitAdvance(); // 等待所有线程到达第一阶段
                System.out.println("Thread " + threadId + " finished phase 1");
                phaser.arriveAndAwaitAdvance(); // 等待所有线程到达第二阶段
                System.out.println("Thread " + threadId + " finished phase 2");
                phaser.arriveAndDeregister(); // 注销线程
            }).start();
        }
    }
}

在这个例子中，我们重写了 onAdvance() 方法，在每个阶段结束后打印阶段数和注册的参与者数量。当所有线程都注销后，onAdvance() 方法返回 true，Phaser 被终止。

4.3 Phaser 的层次结构

Phaser 允许创建层次结构的 Phaser。这可以用于将复杂的任务分解为更小的子任务，并分别协调每个子任务的执行。

例如，我们可以创建一个根 Phaser，然后为每个子任务创建一个子 Phaser。每个子 Phaser 负责协调其子任务的执行。

import java.util.concurrent.Phaser;

public class PhaserHierarchyExample {
    public static void main(String[] args) {
        Phaser rootPhaser = new Phaser(1); // 根 Phaser, 注册 main 线程
        Phaser childPhaser1 = new Phaser(rootPhaser, 2); // 子 Phaser 1, 注册 2 个参与者
        Phaser childPhaser2 = new Phaser(rootPhaser, 3); // 子 Phaser 2, 注册 3 个参与者

        System.out.println("Root Phaser: phase=" + rootPhaser.getPhase() + ", registered=" + rootPhaser.getRegisteredParties());
        System.out.println("Child Phaser 1: phase=" + childPhaser1.getPhase() + ", registered=" + childPhaser1.getRegisteredParties());
        System.out.println("Child Phaser 2: phase=" + childPhaser2.getPhase() + ", registered=" + childPhaser2.getRegisteredParties());

        // 创建线程执行子任务 1
        for (int i = 0; i < 2; i++) {
            new Thread(() -> {
                System.out.println("Child Thread 1 started, phase=" + childPhaser1.getPhase());
                childPhaser1.arriveAndAwaitAdvance();
                System.out.println("Child Thread 1 finished phase 1, phase=" + childPhaser1.getPhase());
                childPhaser1.arriveAndDeregister();
            }).start();
        }

        // 创建线程执行子任务 2
        for (int i = 0; i < 3; i++) {
            new Thread(() -> {
                System.out.println("Child Thread 2 started, phase=" + childPhaser2.getPhase());
                childPhaser2.arriveAndAwaitAdvance();
                System.out.println("Child Thread 2 finished phase 1, phase=" + childPhaser2.getPhase());
                childPhaser2.arriveAndDeregister();
            }).start();
        }

        // Main 线程等待所有子任务完成
        System.out.println("Main thread waiting for children, phase=" + rootPhaser.getPhase());
        rootPhaser.arriveAndAwaitAdvance();
        System.out.println("Main thread finished waiting, phase=" + rootPhaser.getPhase());
        rootPhaser.arriveAndDeregister();

        System.out.println("Root Phaser terminated: " + rootPhaser.isTerminated());
        System.out.println("Child Phaser 1 terminated: " + childPhaser1.isTerminated());
        System.out.println("Child Phaser 2 terminated: " + childPhaser2.isTerminated());
    }
}

在这个例子中，我们创建了一个根 Phaser 和两个子 Phaser。每个子 Phaser 负责协调其子任务的执行。main 线程等待所有子任务完成后才继续执行。

五、Phaser 与 CountDownLatch、CyclicBarrier 的比较

Phaser、CountDownLatch 和 CyclicBarrier 都是 Java 并发包中用于线程同步的工具。它们之间的区别如下：

特性	CountDownLatch	CyclicBarrier	Phaser
主要用途	允许一个或多个线程等待其他线程完成操作	允许一组线程互相等待，直到所有线程都到达一个公共屏障点	用于协调多个线程分阶段完成任务
计数器	只能递减，减到 0 后不可重置	计数器可以重置	阶段数可以递增，参与者可以动态注册和注销
灵活性	较低	较高	极高
适用场景	一次性事件的同步	循环执行的任务的同步	阶段性任务的同步，参与者数量动态变化的情况

• CountDownLatch： 适用于一次性事件的同步。它允许一个或多个线程等待其他线程完成操作。当计数器减到 0 时，所有等待的线程都会被唤醒。CountDownLatch 的计数器只能递减，减到 0 后不可重置。
• CyclicBarrier： 适用于循环执行的任务的同步。它允许一组线程互相等待，直到所有线程都到达一个公共屏障点。当所有线程都到达屏障点后，屏障会被重置，线程可以继续执行下一轮任务。
• Phaser： 适用于阶段性任务的同步。它允许线程动态地注册和注销，并提供了一个 onAdvance() 方法，允许我们在每个阶段结束后执行一些自定义的操作。Phaser 的灵活性最高，可以适应各种复杂的同步场景。

六、使用 Phaser 的注意事项

• 避免死锁： 在使用 Phaser 时，需要注意避免死锁。例如，如果一个线程在等待其他线程到达某个阶段时发生异常，可能会导致死锁。
• 合理设置参与者数量： 在创建 Phaser 对象时，需要合理设置参与者数量。如果参与者数量设置不正确，可能会导致程序无法正常工作。
• 正确处理异常： 在使用 Phaser 的方法时，需要正确处理可能抛出的异常。例如，arriveAndAwaitAdvance() 方法可能会抛出 InterruptedException 异常。
• 避免过度同步： 过度同步会降低程序的性能。在使用 Phaser 时，应该只在必要的时候进行同步。

七、一个更复杂的例子：模拟数据处理管道

假设我们有一个数据处理管道，包含三个阶段：数据读取、数据清洗和数据写入。每个阶段由多个线程并行执行。我们可以使用 Phaser 来协调这些线程的执行。

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Phaser;

public class DataProcessingPipeline {

    private static final int NUM_THREADS = 5;
    private static final int DATA_SIZE = 20;

    public static void main(String[] args) {
        Phaser phaser = new Phaser(1); // 注册 main 线程
        List<Integer> data = generateData(DATA_SIZE);
        List<Integer> processedData = new ArrayList<>(DATA_SIZE);

        // 创建线程执行数据读取阶段
        for (int i = 0; i < NUM_THREADS; i++) {
            final int threadId = i;
            phaser.register(); // 注册工作线程
            new Thread(() -> {
                try {
                    System.out.println("Reader " + threadId + " started");
                    List<Integer> readData = readData(data, threadId, NUM_THREADS);
                    System.out.println("Reader " + threadId + " read " + readData.size() + " elements");

                    // 等待所有 reader 完成
                    phaser.arriveAndAwaitAdvance();
                    System.out.println("Reader " + threadId + " finished reading, phase = " + phaser.getPhase());

                    // 数据清洗阶段
                    List<Integer> cleanedData = cleanData(readData, threadId);

                    // 等待所有 cleaner 完成
                    phaser.arriveAndAwaitAdvance();
                    System.out.println("Cleaner " + threadId + " finished cleaning, phase = " + phaser.getPhase());

                    // 数据写入阶段
                    writeData(cleanedData, processedData, threadId);

                    // 等待所有 writer 完成
                    phaser.arriveAndAwaitAdvance();
                    System.out.println("Writer " + threadId + " finished writing, phase = " + phaser.getPhase());

                } finally {
                    phaser.arriveAndDeregister(); // 注销线程
                    System.out.println("Thread " + threadId + " deregistered, phase = " + phaser.getPhase());
                }
            }).start();
        }

        // 等待所有线程完成
        phaser.arriveAndAwaitAdvance();
        System.out.println("All threads finished, phase = " + phaser.getPhase());

        System.out.println("Processed data size: " + processedData.size());
        phaser.arriveAndDeregister(); // 注销 main 线程
        System.out.println("Main thread deregistered, phase = " + phaser.getPhase());
        System.out.println("Phaser is terminated: " + phaser.isTerminated());
    }

    private static List<Integer> generateData(int size) {
        List<Integer> data = new ArrayList<>(size);
        Random random = new Random();
        for (int i = 0; i < size; i++) {
            data.add(random.nextInt(100));
        }
        return data;
    }

    private static List<Integer> readData(List<Integer> data, int threadId, int numThreads) {
        int chunkSize = data.size() / numThreads;
        int start = threadId * chunkSize;
        int end = (threadId == numThreads - 1) ? data.size() : (threadId + 1) * chunkSize;
        return data.subList(start, end);
    }

    private static List<Integer> cleanData(List<Integer> data, int threadId) {
        List<Integer> cleanedData = new ArrayList<>(data.size());
        for (Integer value : data) {
            cleanedData.add(value * 2); // 简单的数据清洗：乘以 2
        }
        return cleanedData;
    }

    private static void writeData(List<Integer> data, List<Integer> processedData, int threadId) {

        synchronized (processedData) { // 确保多线程写入的线程安全
            processedData.addAll(data); // 简单的将数据添加到 processedData
        }
    }
}

这个例子模拟了一个简单的数据处理管道，其中包含数据读取、数据清洗和数据写入三个阶段。每个阶段由多个线程并行执行。Phaser 用于协调这些线程的执行，确保每个阶段都在所有线程完成之前不会进入下一个阶段。

八、总结说明

Phaser 类是 Java 并发包中一个强大的工具，它提供了一种灵活的机制来协调多个线程的阶段性任务。通过动态注册和注销线程、重写 onAdvance() 方法以及创建层次结构的 Phaser，我们可以轻松地解决各种复杂的线程同步问题。理解 Phaser 的原理和使用方法对于编写高效、可靠的并发程序至关重要。在选择同步工具时，需要根据具体的应用场景选择最合适的工具。CountDownLatch 用于简单的一次性同步，CyclicBarrier 用于循环任务的同步，而 Phaser 则提供了最大的灵活性，适用于各种复杂的阶段性任务。