并发终极杀器一之神秘的Phaser

已剪辑自: https://www.jianshu.com/p/1517a379fb91

前言

Java7为解决线程同步的问题除了提供锁机制外，增加了Semphore、Exchanger、Phaser、CountDownLatch、CyclicBarrier四个工具类来解决多线程多任务同步的情景，本章介绍Phaser它也被称为“阶段器”，可以用它来控制多线程分阶段共同完成的问题，本章后面会介绍一个具体的例子。

常用方法

这里我们将Phaser常用的方法进行分类介绍

1. 设定任务数（这是官方解释，简单的讲比如我们指定任务数为3，那3个线程都运行到指定地方他们才会分别继续往下执行）

\批量设置任务数
public int bulkRegister(int parties)
\动态增加1个任务数
public int register()
\减少一个任务数
public int arriveAndDeregister()

1. 到达等待

\线程执行到此开始等待，满足条件(任务数满足)则继续执行
public int arriveAndAwaitAdvance() 
\ 线程执行到此不用等待继续执行，但会让任务数加1，然后会重置任务数。
public int arrive()
\传入的phase表示翻越几个屏障，如果传入的等于getPhase则等待
public int awaitAdvance(int phase)
\可中断的awaitAdvance
public int awaitAdvanceInterruptibly(int phase)；
\可规定时间指定栏数未变，则抛出异常
public int awaitAdvanceInterruptibly(int phase,
                                         long timeout, TimeUnit unit)

1. 得到栏数以及任务数

\当前执行到第几个屏障
public final int getPhase()
\得到注册的任务数
public int getRegisteredParties()
\已经执行的任务数
public int getArrivedParties()
\还未执行的任务数
public int getUnarrivedParties()

1. 通过屏障时调用

\该方法返回true则直接通过屏障 false则屏障继续工作，要想实现自己的逻辑代码需要复写此方法
protected boolean onAdvance(int phase, int registeredParties)

1. 关闭屏障

\取消屏障
public void forceTermination()
\phaser是否销毁
public boolean isTerminated()

小例子

有10名运动员参加一场百米竞赛，分为3阶段：第一阶段所有选手到达赛场、第二阶段选手在起跑线就位、完成比赛。

public class PharseDemo {

    public static void main(String[] args) throws InterruptedException {
        //10名运动员，每名运动员在同一时间点的开始
        Phaser pharse = new Phaser(10) {
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                if(phase == 0) {
                    System.out.println("所有人员已经到赛场");
                }else if(phase == 1) {
                    System.out.println("所有人员准备完毕，开始比赛");
                }else if(phase == 2) {
                    System.out.println("比赛结束");
                }
                return super.onAdvance(phase, registeredParties);
            }
        };
        //创建10个线程
        Runner run = new Runner(pharse);
        Thread[] thread  = new Thread[10];
        for (int i = 0; i < 10; i++) {
            thread[i] = new Thread(run);
        }
        Thread.sleep(3000);
        for (int i = 0; i < 10; i++) {
            thread[i].start();
        }
    }
    
}

class Runner implements Runnable{

    private Phaser phaser;
    Random random = new Random();
    
    public Runner(Phaser phaser) {
        super();
        this.phaser = phaser;
    }



    @Override
    public void run() {
        try {
            //第一阶段  到达赛场 每个运动员情况不一样
            Thread.sleep(random.nextInt(3000));
            System.out.println(Thread.currentThread().getName()+"已经到达赛场");
            phaser.arriveAndAwaitAdvance();
            //第二阶段   开始准备 
            Thread.sleep(random.nextInt(3000));
            System.out.println(Thread.currentThread().getName()+"已经准备好");
            phaser.arriveAndAwaitAdvance();
            //第二阶段   到达终点
            Thread.sleep(random.nextInt(3000));
            System.out.println(Thread.currentThread().getName()+"已经到达终点");
            phaser.arriveAndAwaitAdvance();
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }
    
}

代码中注释很详细，这里就不赘述。通过Phaser的demo可以看出Phaser的诸多优点：动态增加任务数、得到程序执行刀第几阶段等等......，有了以上支持在多线程任务分阶段执行时就能更好的解决同步的问题了。