Java并發(fā)工具類Phaser詳解

更新時間：2023年11月30日 08:58:44 作者：啊幾

這篇文章主要介紹了Java并發(fā)工具類Phaser詳解,Phaser(階段協(xié)同器)是一個Java實現(xiàn)的并發(fā)工具類,用于協(xié)調(diào)多個線程的執(zhí)行,它提供了一些方便的方法來管理多個階段的執(zhí)行,可以讓程序員靈活地控制線程的執(zhí)行順序和階段性的執(zhí)行,需要的朋友可以參考下

前言

Phaser（階段協(xié)同器）是一個Java實現(xiàn)的并發(fā)工具類，用于協(xié)調(diào)多個線程的執(zhí)行。

它提供了一些方便的方法來管理多個階段的執(zhí)行，可以讓程序員靈活地控制線程的執(zhí)行順序和階段性的執(zhí)行。

Phaser可以被視為CyclicBarrier和CountDownLatch的進化版，它能夠自適應地調(diào)整并發(fā)線程數(shù)，可以動態(tài)地增加或減少參與線程的數(shù)量。

所以Phaser特別適合使用在重復執(zhí)行或者重用的情況。

在這里插入圖片描述

常用API

構(gòu)造方法

Phaser(): 參與任務數(shù)0
Phaser(int parties) :指定初始參與任務數(shù)
Phaser(Phaser parent) :指定parent階段器，子對象作為一個整體加入parent對象，當子對象中沒有參與者時，會自動從parent對象解除注冊
Phaser(Phaser parent，int parties) : 集合上面兩個方法

增減參與任務數(shù)方法

int register() 增加一個任務數(shù)，返回當前階段號。
int bulkRegister(int parties) 增加指定任務個數(shù)，返回當前階段號。
int arriveAndDeregister() 減少一個任務數(shù)，返回當前階段號。

到達、等待方法

int arrive() 到達(任務完成)，返回當前階段號。
int arriveAndAwaitAdvance() 到達后等待其他任務到達，返回到達階段號。
int awaitAdvance(int phase) 在指定階段等待(必須是當前階段才有效)
int awaitAdvanceInterruptibly(int phase) 階段到達觸發(fā)動作
int awaitAdvanceInterruptiBly(int phase，long timeout，TimeUnit unit)
protected boolean onAdvance(int phase，int registeredParties)類似CyclicBarrier的觸發(fā)命令，通過重寫該方法來增加階段到達動作，該方法返回true將終結(jié)Phaser對象。

Phaser使用

多線程批量處理數(shù)據(jù)

public class PhaserBatchProcessorDemo {

    private final List<String> data;
    private final int batchsize;//一次處理多少數(shù)據(jù)
    private final int threadCount;//處理的線程數(shù)
    private final Phaser phaser;
    private final List<String> processedData;

    public PhaserBatchProcessorDemo(List<String> data,int batchsize,int threadCount){
        this.data = data;
        this.batchsize = batchsize;
        this.threadCount = threadCount;
        this.phaser = new Phaser(1);
        //this.phaser = new Phaser();
        this.processedData = new ArrayList<>();
    }

    public void process() throws InterruptedException {
        for(int i = 0;i<threadCount;i++){
            System.out.println("phaser.register():"+phaser.register());
            new Thread(new BatchProcessor(i)).start();
            Thread.sleep(50);
        }
        phaser.arriveAndDeregister();//主線程執(zhí)行結(jié)束
        System.out.println("結(jié)束");
    }

    private class BatchProcessor implements Runnable{
        private final int threadIndex;
        public BatchProcessor(int threadIndex){this.threadIndex = threadIndex;}

        @Override
        public void run() {
            int index = 0;
            while(true){
                //所有線程都到達這個點之前會阻塞
                System.out.println("線程"+threadIndex+"phaser.arriveAndAwaitAdvance1():");
                phaser.arriveAndAwaitAdvance();

                //從未處理數(shù)據(jù)中找到一個可以處理的批次
                List<String> batch = new ArrayList<>();
                synchronized (data){
                    while (index < data.size()&&batch.size()<batchsize){
                        String d = data.get(index);
                        if(!processedData.contains(d)){
                            batch.add(d);
                            processedData.add(d);
                        }
                        index++;
                    }
                }
                //處理數(shù)據(jù)
                for(String d:batch){
                    System.out.println("線程"+threadIndex+"處理數(shù)據(jù)"+d);
                }
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
                //所有數(shù)據(jù)都處理完當前批次之前會阻塞
                System.out.println("線程"+threadIndex+"phaser.arriveAndAwaitAdvance2():");
                phaser.arriveAndAwaitAdvance();
                //所有線程都處理完當前批次并且未處理數(shù)據(jù)已經(jīng)處理完之前會阻塞
                if(batch.isEmpty()||index >= data.size()){
                    System.out.println("線程"+threadIndex+"phaser.arriveAndDeregister()"+phaser.arriveAndDeregister());
                    break;
                }
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        List<String> data = new ArrayList<>();
        for(int i = 1;i<=15;i++){
            data.add(String.valueOf(i));
        }

        PhaserBatchProcessorDemo processor = new PhaserBatchProcessorDemo(data,4,3);
        processor.process();
    }
}

/**
 * phaser.register():0
 * 線程0phaser.arriveAndAwaitAdvance1():
 * phaser.register():0
 * 線程1phaser.arriveAndAwaitAdvance1():
 * phaser.register():0
 * 線程2phaser.arriveAndAwaitAdvance1():
 * 結(jié)束
 * 線程2處理數(shù)據(jù)9
 * 線程2處理數(shù)據(jù)10
 * 線程1處理數(shù)據(jù)5
 * 線程1處理數(shù)據(jù)6
 * 線程1處理數(shù)據(jù)7
 * 線程1處理數(shù)據(jù)8
 * 線程2處理數(shù)據(jù)11
 * 線程2處理數(shù)據(jù)12
 * 線程0處理數(shù)據(jù)1
 * 線程0處理數(shù)據(jù)2
 * 線程0處理數(shù)據(jù)3
 * 線程0處理數(shù)據(jù)4
 * 線程2phaser.arriveAndAwaitAdvance2():
 * 線程0phaser.arriveAndAwaitAdvance2():
 * 線程1phaser.arriveAndAwaitAdvance2():
 * 線程2phaser.arriveAndAwaitAdvance1():
 * 線程1phaser.arriveAndAwaitAdvance1():
 * 線程0phaser.arriveAndAwaitAdvance1():
 * 線程0處理數(shù)據(jù)13
 * 線程0處理數(shù)據(jù)14
 * 線程0處理數(shù)據(jù)15
 * 線程0phaser.arriveAndAwaitAdvance2():
 * 線程1phaser.arriveAndAwaitAdvance2():
 * 線程2phaser.arriveAndAwaitAdvance2():
 * 線程1phaser.arriveAndDeregister()4
 * 線程2phaser.arriveAndDeregister()4
 * 線程0phaser.arriveAndDeregister()4
 */

這里提出一個問題：為什么要給主線程也注冊呢？如果不給主線程注冊會怎么樣呢？

在這里插入圖片描述

這里就要提及register() 增加任務數(shù)量和Phaser()初始化定義任務數(shù)量的區(qū)別：

register()有一個需要注意的點是，如果主線程執(zhí)行速度緩慢的話，很有可能在第一個線程已經(jīng)arrive的時候，第二個線程任務還沒增加，導致第一個線程因為parties只有1，而沒有阻塞等待就進入下一階段了。

如果不給主線程注冊添加任務，運行結(jié)果如下

phaser.register():0
線程0phaser.arriveAndAwaitAdvance1():
線程0處理數(shù)據(jù)1
線程0處理數(shù)據(jù)2
線程0處理數(shù)據(jù)3
線程0處理數(shù)據(jù)4
phaser.register():1
線程1phaser.arriveAndAwaitAdvance1():
phaser.register():1
線程2phaser.arriveAndAwaitAdvance1():
結(jié)束
線程0phaser.arriveAndAwaitAdvance2():
線程0phaser.arriveAndAwaitAdvance1():
線程2處理數(shù)據(jù)5
線程2處理數(shù)據(jù)6
線程2處理數(shù)據(jù)7
線程2處理數(shù)據(jù)8
線程1處理數(shù)據(jù)9
線程1處理數(shù)據(jù)10
線程1處理數(shù)據(jù)11
線程1處理數(shù)據(jù)12
線程2phaser.arriveAndAwaitAdvance2():
線程1phaser.arriveAndAwaitAdvance2():
線程2phaser.arriveAndAwaitAdvance1():
線程1phaser.arriveAndAwaitAdvance1():
線程0處理數(shù)據(jù)13
線程0處理數(shù)據(jù)14
線程0處理數(shù)據(jù)15
線程0phaser.arriveAndAwaitAdvance2():
線程0phaser.arriveAndDeregister()4
線程1phaser.arriveAndAwaitAdvance2():
線程2phaser.arriveAndAwaitAdvance2():
線程2phaser.arriveAndDeregister()5
線程1phaser.arriveAndDeregister()5

而Phaser()初始化就定義了parties，會讓所有線程都必須到達之前都阻塞才能進入下一階段。

給主線程也增加一個任務的目的就在于此 如果主線程也有任務，就算主線程執(zhí)行緩慢，第一個線程也必須阻塞等待主線程在第一階段之前，把所有線程都start()啟動。

階段性任務：模擬伙伴出游

public class PhaserDemo {
    public static void main(String[] args) {
        final Phaser phaser = new Phaser(){
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                //參與者數(shù)量，去除主線程
                int persons = registeredParties - 1;
                switch (phase){
                    case 0:
                        System.out.println("大家都到佘山站了，可以出發(fā)去佘山了，人數(shù)："+persons);
                        break;
                    case 1:
                        System.out.println("大家都到佘山了，出發(fā)去爬山，人數(shù)："+persons);
                        break;
                    case 2:
                        System.out.println("大家都到山頂了，開始休息，人數(shù)："+persons);
                        break;

                }
                //判斷是否只剩下一個主線程，如果是，返回true,代表終止
                return registeredParties ==1;
            }
        };

        phaser.register();
        final PersonTask personTask = new PersonTask();
        //3個全程參加的伙伴
        for(int i = 0;i<3;i++){

            phaser.register();
            new Thread(()->{
                try{
                    personTask.step1Task();
                    phaser.arriveAndAwaitAdvance();

                    personTask.step2Task();
                    phaser.arriveAndAwaitAdvance();

                    personTask.step3Task();
                    phaser.arriveAndAwaitAdvance();

                    personTask.step4Task();
                    phaser.arriveAndDeregister();


                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }).start();
        }
        

        //兩個在山腰半路返回
        for(int i = 0;i<2;i++){
            phaser.register();
            new Thread(()->{
                try{
                    personTask.step1Task();
                    phaser.arriveAndAwaitAdvance();

                    personTask.step2Task();
                    phaser.arriveAndAwaitAdvance();

                    personTask.step3Task();
                    System.out.println("伙伴【"+Thread.currentThread().getName()+"】中途山腰返回");
                    phaser.arriveAndDeregister();

                }catch (InterruptedException e){
                    e.printStackTrace();
                }
            }).start();
        }

        while (!phaser.isTerminated()) {
            int phase = phaser.arriveAndAwaitAdvance();
            if (phase == 2) {
                //兩個在佘山直接會合
                for(int i = 0;i<2;i++){
                    phaser.register();
                    new Thread(()->{
                        try{
                            System.out.println("伙伴【"+Thread.currentThread().getName()+"】中途加入");

                            personTask.step3Task();
                            phaser.arriveAndAwaitAdvance();

                            personTask.step4Task();
                            phaser.arriveAndDeregister();

                        }catch (InterruptedException e){
                            e.printStackTrace();
                        }
                    }).start();
                }
            }
        }
    }

    static final Random random = new Random();

    static class PersonTask{
        public void step1Task() throws InterruptedException {
            String person = "伙伴【"+Thread.currentThread().getName()+"】";
            System.out.println(person+"從家出發(fā)了......");
            Thread.sleep(random.nextInt(5000));
            System.out.println(person+"到達佘山站");
        }

        public void step2Task() throws InterruptedException {
            String person = "伙伴【"+Thread.currentThread().getName()+"】";
            System.out.println(person+"出發(fā)去佘山");
            Thread.sleep(random.nextInt(5000));
            System.out.println(person+"到達佘山");
        }

        public void step3Task() throws InterruptedException {
            String person = "伙伴【"+Thread.currentThread().getName()+"】";
            System.out.println(person+"出發(fā)去爬山");
            Thread.sleep(random.nextInt(5000));
            System.out.println(person+"到達山頂");
        }

        public void step4Task() throws InterruptedException {
            String person = "伙伴【"+Thread.currentThread().getName()+"】";
            System.out.println(person+"開始休息");
            Thread.sleep(random.nextInt(5000));
            System.out.println(person+"休息結(jié)束，下山回家");
        }
    }
}

/**
 * 伙伴【Thread-3】從家出發(fā)了......
 * 伙伴【Thread-1】從家出發(fā)了......
 * 伙伴【Thread-2】從家出發(fā)了......
 * 伙伴【Thread-4】從家出發(fā)了......
 * 伙伴【Thread-0】從家出發(fā)了......
 * 伙伴【Thread-1】到達佘山站
 * 伙伴【Thread-4】到達佘山站
 * 伙伴【Thread-3】到達佘山站
 * 伙伴【Thread-0】到達佘山站
 * 伙伴【Thread-2】到達佘山站
 * 大家都到佘山站了，可以出發(fā)去佘山了，人數(shù)：5
 * 伙伴【Thread-3】出發(fā)去佘山
 * 伙伴【Thread-2】出發(fā)去佘山
 * 伙伴【Thread-4】出發(fā)去佘山
 * 伙伴【Thread-1】出發(fā)去佘山
 * 伙伴【Thread-0】出發(fā)去佘山
 * 伙伴【Thread-1】到達佘山
 * 伙伴【Thread-4】到達佘山
 * 伙伴【Thread-3】到達佘山
 * 伙伴【Thread-0】到達佘山
 * 伙伴【Thread-2】到達佘山
 * 大家都到佘山了，出發(fā)去爬山，人數(shù)：5
 * 伙伴【Thread-1】出發(fā)去爬山
 * 伙伴【Thread-4】出發(fā)去爬山
 * 伙伴【Thread-0】出發(fā)去爬山
 * 伙伴【Thread-2】出發(fā)去爬山
 * 伙伴【Thread-3】出發(fā)去爬山
 * 伙伴【Thread-6】中途加入
 * 伙伴【Thread-6】出發(fā)去爬山
 * 伙伴【Thread-5】中途加入
 * 伙伴【Thread-5】出發(fā)去爬山
 * 伙伴【Thread-2】到達山頂
 * 伙伴【Thread-5】到達山頂
 * 伙伴【Thread-0】到達山頂
 * 伙伴【Thread-4】到達山頂
 * 伙伴【Thread-4】中途山腰返回
 * 伙伴【Thread-6】到達山頂
 * 伙伴【Thread-3】到達山頂
 * 伙伴【Thread-3】中途山腰返回
 * 伙伴【Thread-1】到達山頂
 * 大家都到山頂了，開始休息，人數(shù)：5
 * 伙伴【Thread-5】開始休息
 * 伙伴【Thread-2】開始休息
 * 伙伴【Thread-6】開始休息
 * 伙伴【Thread-0】開始休息
 * 伙伴【Thread-1】開始休息
 * 伙伴【Thread-2】休息結(jié)束，下山回家
 * 伙伴【Thread-0】休息結(jié)束，下山回家
 * 伙伴【Thread-1】休息結(jié)束，下山回家
 * 伙伴【Thread-5】休息結(jié)束，下山回家
 * 伙伴【Thread-6】休息結(jié)束，下山回家
 */