j.u.c.Phaser demo

phaser.log

[Slow consumer    ] about to wait for phase [0] completion
[Producer 1       ] working in lane1 finished phase [0]
[Slower producer 2] working in lane2 finished phase [0]
[Slow consumer    ] read [lane1-answer-0] & [lane2-answer-0] from phase [0]
[Producer 1       ] working in lane1 finished phase [1]
[Slower producer 2] working in lane2 finished phase [1]
[Producer 1       ] working in lane1 finished phase [2]
[Slow consumer    ] about to wait for phase [1] completion
[Slow consumer    ] read [lane1-answer-1] & [lane2-answer-1] from phase [1]
[Slower producer 2] working in lane2 finished phase [2]
[Producer 1       ] working in lane1 finished phase [3]
[Slower producer 2] working in lane2 finished phase [3]
[Producer 1       ] working in lane1 finished phase [4]
[Slow consumer    ] about to wait for phase [2] completion
[Slow consumer    ] read [lane1-answer-2] & [lane2-answer-2] from phase [2]
[Slow consumer    ] read [lane1-answer-3] & [lane2-answer-3] from phase [3]
[Slower producer 2] working in lane2 finished phase [4]
[Producer 1       ] working in lane1 finished phase [5]
[Slower producer 2] working in lane2 finished phase [5]
[Producer 1       ] working in lane1 finished phase [6]
[Slow consumer    ] about to wait for phase [4] completion
[Slow consumer    ] read [lane1-answer-4] & [lane2-answer-4] from phase [4]
[Slow consumer    ] read [lane1-answer-5] & [lane2-answer-5] from phase [5]
[Slower producer 2] working in lane2 finished phase [6]
[Producer 1       ] working in lane1 finished phase [7]
[Slower producer 2] working in lane2 finished phase [7]
[Producer 1       ] working in lane1 finished phase [8]
[Slow consumer    ] about to wait for phase [6] completion
[Slow consumer    ] read [lane1-answer-6] & [lane2-answer-6] from phase [6]
[Slow consumer    ] read [lane1-answer-7] & [lane2-answer-7] from phase [7]
[Slower producer 2] working in lane2 finished phase [8]
[Producer 1       ] working in lane1 finished phase [9]
[Slower producer 2] working in lane2 finished phase [9]
[Slow consumer    ] about to wait for phase [8] completion
[Slow consumer    ] read [lane1-answer-8] & [lane2-answer-8] from phase [8]
[Slow consumer    ] read [lane1-answer-9] & [lane2-answer-9] from phase [9]

PhaserDemo.java

import java.util.concurrent.Phaser;
import java.util.concurrent.atomic.AtomicReferenceArray;

/*
* Author: Ashwin Jayaprakash
*/
public class PhaserDemo {
    public static void main(String[] args) {
        final int workers = 2;
        final int workLength = 10;

        final Phaser phaser = new Phaser(workers + 1);
        final AtomicReferenceArray<String> lane1 = new AtomicReferenceArray<String>(new String[workLength]);
        final AtomicReferenceArray<String> lane2 = new AtomicReferenceArray<String>(new String[workLength]);

        new Thread("Producer 1") {
            @Override
            public void run() {
                for (int i = 0; i < workLength; i++) {
                    $sleep(20);

                    lane1.set(i, "lane1-answer-" + i);

                    System.out.printf("[%-17s] working in lane1 finished phase [%d]%n",
                            Thread.currentThread().getName(), phaser.getPhase());

                    phaser.arriveAndAwaitAdvance();
                }
            }
        }.start();

        new Thread("Slower producer 2") {
            @Override
            public void run() {
                for (int i = 0; i < workLength; i++) {
                    $sleep(40);

                    lane2.set(i, "lane2-answer-" + i);

                    System.out.printf("[%-17s] working in lane2 finished phase [%d]%n",
                            Thread.currentThread().getName(), phaser.getPhase());

                    phaser.arriveAndAwaitAdvance();
                }
            }
        }.start();

        new Thread("Slow consumer") {
            @Override
            public void run() {
                for (int start = 0; start < workLength; ) {
                    System.out.printf("[%-17s] about to wait for phase [%d] completion%n",
                            Thread.currentThread().getName(), start);

                    int phaseInProgress = phaser.awaitAdvance(start);

                    //Read all the way up to the most recent completed phases.
                    for (int i = start; i < phaseInProgress; i++) {
                        System.out.printf("[%-17s] read [%s] & [%s] from phase [%d]%n",
                                Thread.currentThread().getName(), lane1.get(i), lane2.get(i), i);
                    }

                    start = phaseInProgress;

                    $sleep(80);
                }
            }
        }.start();

        phaser.arriveAndDeregister();
    }

    private static void $sleep(long millis) {
        try {
            Thread.sleep(millis);
        }
        catch (InterruptedException e) {
        }
    }
}