Java并发编程：Phaser 多阶段任务并行执行的利器（使用范例）

爱被打了一巴掌 2022-02-26 16:42 283阅读 0赞

Phaser 类是 JUC 包下的并发控制类，适用于多线程分阶段有序的并行执行任务的场景，类似于CountDownLatch、CyclicBarrier 的功能，但是 Phaser 类更加灵活和方便。本篇文章的目的是通俗易懂的讲解 Phaser 的概念以及如何使用 Phaser，便于大家快速理解 Phaser 。

### 1 概念讲解 ###

> 阶段=0，线程=thread2，时间=2019-03-30 14:16:42 正在执行  
> 阶段=0，线程=thread1，时间=2019-03-30 14:16:42 正在执行  
> 阶段=0，线程=thread0，时间=2019-03-30 14:16:42 正在执行  
> 阶段=0，线程=thread2，时间=2019-03-30 14:16:43 完成

> 阶段=1，线程=thread2，时间=2019-03-30 14:16:43 正在执行  
> 阶段=1，线程=thread1，时间=2019-03-30 14:16:43 正在执行  
> 阶段=1，线程=thread0，时间=2019-03-30 14:16:43 正在执行  
> 阶段=1，线程=thread0，时间=2019-03-30 14:16:44 完成

> 阶段=2，线程=thread2，时间=2019-03-30 14:16:44 正在执行  
> 阶段=2，线程=thread0，时间=2019-03-30 14:16:44 正在执行  
> 阶段=2，线程=thread1，时间=2019-03-30 14:16:44 正在执行  
> 阶段=2，线程=thread1，时间=2019-03-30 14:16:45 完成

多阶段：分为阶段0、阶段1、阶段2  
多线程：每个阶段中有线程0、线程1、线程2三个线程  
有序性：阶段0、阶段1、阶段2按照顺序同步执行  
并行执行：在每个阶段中，线程0、线程1、线程2在并行执行任务

### 2 代码示例 ###

#### 2.1 定义 Phaser 类 ####

首先我们先定义一个 Phaser 类，主要目的是重写 onAdvance 方法，来控制 Phaser 是否继续执行后面的阶段。onAdvance 的实现可以是多种多样的，我们列举两个实现：

示例一：通过注册的参与者数量进行控制，这也是 JDK 中的默认实现

/**
         * onAdvance 是一个回调方法，用来控制 Phaser 是否继续执行后面的阶段。我们可以通过两个入参来控制方法的返回结果。
         * @param phase Phaser 当前所处的阶段
         * @param registeredParties 已经注册的参与者（线程）数量
         * @return 返回值为 ture，则 Phaser 停止执行后续阶段，返回值为 false，则 Phaser 继续执行后续阶段
         */
        protected boolean onAdvance(int phase, int registeredParties) {
        	// 如果已经注册的参与者（线程）数量为0，则停止执行后续阶段
            return registeredParties == 0;
        }

示例二：通过阶段进行控制

// 如果Phaser 所处的阶段为0或1，返回 false继续执行后续阶段
        // 否则停止执行后续阶段
        protected boolean onAdvance(int phase, int registeredParties) {
            switch (phase) {
                case 0:
                    firstPhaser();
                    return false;
    			case 1:
                    secondPhaser();
                    return false;
                default:
                    return true;
            }
        }
        }

了解了 onAdvance 方法后，我们定义一个 Phaser 的派生类 TaskPhaser，代码如下：

package org.learn.juc;
    
    import java.text.DateFormat;
    import java.text.SimpleDateFormat;
    import java.util.Date;
    import java.util.concurrent.Phaser;
    
    public class TaskPhaser extends Phaser {
        private final static DateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    
        TaskPhaser(int parties) {
            super(parties);
        }
    
        /**
         * onAdvance 是一个回调方法，用来控制 Phaser 是否继续执行，我们可以通过两个入参来控制该方法的返回结果
         * 该方法返回 ture，则 Phaser 停止执行
         * 该方法返回 false，则 Phaser 继续执行
         *
         * @param phase             当前所处的阶段（phase）
         * @param registeredParties 已经注册的参与者
         * @return
         */
        @Override
        protected boolean onAdvance(int phase, int registeredParties) {
            switch (phase) {
                case 0:
                    firstPhaser();
                    return false;
                case 1:
                    secondPhaser();
                    return false;
                case 2:
                    thirdPhaser();
                    return false;
                default:
                    return true;
            }
        }
    
        private void firstPhaser() {
            System.out.println(String.format("阶段=0，线程=%s，时间=%s 完成 \n", Thread.currentThread().getName(), sdf.format(new Date())));
        }
    
        private void secondPhaser() {
            System.out.println(String.format("阶段=1，线程=%s，时间=%s 完成 \n", Thread.currentThread().getName(), sdf.format(new Date())));
        }
    
        private void thirdPhaser() {
            System.out.println(String.format("阶段=2，线程=%s，时间=%s 完成 \n", Thread.currentThread().getName(), sdf.format(new Date())));
        }
    }

#### 2.2 定义 Task 类（线程类） ####

定义 Task 类（线程类），通过构造函数将 TaskPhaser 赋值给 Task 类，在该类中通过 Phaser 的 arriveAndAwaitAdvance 方法实现分阶段同步执行任务。arriveAndAwaitAdvance 方法的含义是到达并等待继续前进，继续前进的条件是 Phaser 中的全部参与者都已经到达当前阶段的同步点。比如 Phaser 有 5 个参与者，并且当前所处阶段为0，那么只有五个参与者都到达阶段0的同步点，则根据 onAdvance 来决定是否继续执行下一阶段。

Task 类（线程类）代码如下：

package org.learn.juc;
    
    import java.text.DateFormat;
    import java.text.SimpleDateFormat;
    import java.util.Date;
    
    public class Task implements Runnable {
        private final static DateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    
        private TaskPhaser phaser;
    
        Task(TaskPhaser phaser) {
            this.phaser = phaser;
        }
    
        @Override
        public void run() {
            /// 第一阶段
            firstPhaseWork();
            phaser.arriveAndAwaitAdvance();
    
            /// 第二阶段
            secondPhaseWork();
            phaser.arriveAndAwaitAdvance();
    
            /// 第三阶段
            thirdPhaseWork();
            phaser.arriveAndAwaitAdvance();
        }
    
        private void firstPhaseWork() {
            System.out.println(String.format("阶段=%s，线程=%s，时间=%s 正在执行", phaser.getPhase(), Thread.currentThread().getName(), sdf.format(new Date())));
            try {
                Thread.sleep(1000);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        private void secondPhaseWork() {
            System.out.println(String.format("阶段=%s，线程=%s，时间=%s 正在执行", phaser.getPhase(), Thread.currentThread().getName(), sdf.format(new Date())));
            try {
                Thread.sleep(1000);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        private void thirdPhaseWork() {
            System.out.println(String.format("阶段=%s，线程=%s，时间=%s 正在执行", phaser.getPhase(), Thread.currentThread().getName(), sdf.format(new Date())));
            try {
                Thread.sleep(1000);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

#### 2.3 执行代码 ####

main 方法代码如下：

package org.learn.juc;
    
    public class PhaserDemo {
        public static void main(String[] args) {
            /// 参与者数量
            int parties = 3;
    
            /// 定义 Phaser 类，并设置参与者数量为3
            TaskPhaser phaser = new TaskPhaser(parties);
    
            Thread thread = null;
            /// 定义3个参与者
            for (int i = 0; i < parties; i++) {
                thread = new Thread(new Task(phaser), "thread" + i);
                thread.start();
            }
        }
    }

查看运行结果：

阶段=0，线程=thread2，时间=2019-03-30 14:51:19 正在执行
    阶段=0，线程=thread0，时间=2019-03-30 14:51:19 正在执行
    阶段=0，线程=thread1，时间=2019-03-30 14:51:19 正在执行
    阶段=0，线程=thread1，时间=2019-03-30 14:51:20 完成 
    
    阶段=1，线程=thread2，时间=2019-03-30 14:51:20 正在执行
    阶段=1，线程=thread0，时间=2019-03-30 14:51:20 正在执行
    阶段=1，线程=thread1，时间=2019-03-30 14:51:20 正在执行
    阶段=1，线程=thread1，时间=2019-03-30 14:51:21 完成 
    
    阶段=2，线程=thread2，时间=2019-03-30 14:51:21 正在执行
    阶段=2，线程=thread1，时间=2019-03-30 14:51:21 正在执行
    阶段=2，线程=thread0，时间=2019-03-30 14:51:21 正在执行
    阶段=2，线程=thread0，时间=2019-03-30 14:51:22 完成

Phaser 类提供了多个方法用来控制参与者的注册、撤销等，在这里先不展开讲解，后续会写一篇关于 Phaser 类的源码分析的文章，会详细的介绍每个方法。文章内容仅代表个人观点，如有不正之处，欢迎批评指正，谢谢大家。