Java如何使用Phaser实现复杂任务协调_Java并发阶段控制模型-java教程-PHP中文网

Java如何使用Phaser实现复杂任务协调_Java并发阶段控制模型

P粉602998670

发布： 2025-11-18 14:27:24

原创

911人浏览过

Phaser适合多阶段线程协同，支持动态注册与注销，通过arriveAndAwaitAdvance实现阶段同步，结合onAdvance定制各阶段逻辑，适用于参与线程数变化的分步并发场景。

java如何使用phaser实现复杂任务协调_java并发阶段控制模型

在Java并发编程中，当需要对多个线程按阶段协同执行时，Phaser 是比 CountDownLatch 或 CyclicBarrier 更灵活的选择。它支持动态注册线程、分阶段同步、以及更复杂的协调逻辑。特别适合用于需要分步执行、每阶段参与线程数可能变化的场景。

Phaser的基本概念与核心机制

Phaser 可以看作是 CountDownLatch 和 CyclicBarrier 的结合体，但它支持可变参与线程数和重复使用。每个线程通过 register() 或 arriveAndAwaitAdvance() 加入阶段同步。

关键方法说明：

arriveAndAwaitAdvance()：通知当前阶段到达，并阻塞等待所有参与者完成
arriveAndDeregister()：到达但不再参与后续阶段（退出Phaser）
register()：动态注册一个新参与者
bulkRegister(n)：批量注册n个参与者
getPhase()：获取当前阶段编号（从0开始）

模拟多阶段任务处理流程

假设有一个数据处理系统，包含三个阶段：数据加载、数据校验、数据导出。不同阶段参与的线程数量可能不同。

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

ImgCleaner

一键去除图片内的任意文字，人物和对象

220

查看详情

public class MultiStageProcessor {
    private static final Phaser phaser = new Phaser(1); // 主线程作为控制器
<pre class='brush:java;toolbar:false;'>public static void main(String[] args) {
    // 阶段0：启动
    System.out.println("系统启动，准备进入第一阶段");

    // 第一阶段：数据加载
    for (int i = 0; i < 3; i++) {
        new Thread(new DataLoader(i)).start();
    }

    phaser.arriveAndAwaitAdvance(); // 等待所有加载器完成
    System.out.println("✅ 第一阶段完成：" + phaser.getPhase());

    // 第二阶段：数据校验（新增校验线程）
    phaser.register(); // 新增一个校验线程
    new Thread(new DataValidator()).start();

    phaser.arriveAndAwaitAdvance();
    System.out.println("✅ 第二阶段完成：" + phaser.getPhase());

    // 第三阶段：数据导出（部分线程退出）
    new Thread(new DataExporter()).start();

    phaser.arriveAndAwaitAdvance();
    System.out.println("✅ 第三阶段完成：" + phaser.getPhase());

    phaser.arriveAndDeregister(); // 主线程退出
}

static class DataLoader implements Runnable {
    private final int id;

    DataLoader(int id) { this.id = id; }

    @Override
    public void run() {
        phaser.register(); // 注册自己
        try {
            System.out.println("加载器 " + id + " 正在加载数据...");
            Thread.sleep(500 + id * 100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            phaser.arriveAndAwaitAdvance();
        }
    }
}

static class DataValidator implements Runnable {
    @Override
    public void run() {
        try {
            System.out.println("校验器正在校验数据...");
            Thread.sleep(600);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            phaser.arriveAndAwaitAdvance();
        }
    }
}

static class DataExporter implements Runnable {
    @Override
    public void run() {
        phaser.register();
        try {
            System.out.println("导出器正在导出数据...");
            Thread.sleep(800);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        } finally {
            phaser.arriveAndAwaitAdvance();
        }
    }
}

登录后复制

}

Phaser的进阶用法与注意事项

Phaser 支持自定义阶段行为，可通过重写 onAdvance(int phase, int registeredParties) 方法实现每阶段结束时的判断逻辑。

Phaser phaser = new Phaser() {
    @Override
    protected boolean onAdvance(int phase, int registeredParties) {
        switch (phase) {
            case 0:
                System.out.println("第一阶段结束，进入校验");
                return false; // 继续下一阶段
            case 1:
                System.out.println("校验完成，准备导出");
                return false;
            case 2:
                System.out.println("所有任务完成！");
                return true; // 返回true则终止Phaser
            default:
                return true;
        }
    }
};

登录后复制

使用时注意：