How to Build a More Flexible Java Phaser: Introducing FunPhaser

Background and Motivation

The standard java.util.concurrent.Phaser is useful for coordinating many asynchronous tasks, but it imposes a hard limit of 65,535 parties ( MAX_PARTIES = 0xffff) and offers a rich API that can be confusing for simple use cases. When handling a variable number of tasks, these constraints can lead to data loss and cumbersome code.

Design Goals for FunPhaser

To overcome these issues, a lightweight custom synchronizer named FunPhaser was created with the following objectives:

Thread‑safe counting of completed tasks.

Thread‑safe tracking of pending tasks.

Simple registration method to add tasks.

Completion method to decrement pending count and increment completed count.

Method to await all tasks finishing.

Method to query the number of completed tasks.

Implementation Details

package com.funtester.frame;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * Custom synchronizer that avoids the 65535‑party limitation of java.util.concurrent.Phaser.
 * Use #done() after a task finishes; when the total count reaches 0, #await() returns.
 */
class FunPhaser extends SourceCode {
    /** Task index – increments on registration, decrements on completion */
    AtomicInteger index;
    /** Total completed task count */
    AtomicInteger taskNum;

    FunPhaser() {
        this.index = new AtomicInteger();
        this.taskNum = new AtomicInteger();
    }

    /** Register a new task */
    def register() {
        this.index.getAndIncrement();
    }

    /** Mark a task as done */
    def done() {
        this.index.getAndDecrement();
        this.taskNum.getAndIncrement();
    }

    /** Wait until all tasks are finished */
    def await() {
        waitFor { index.get() == 0 }
    }

    /** Get the number of completed tasks */
    int queryTaskNum() {
        return taskNum.get();
    }
}

Demo Usage

import com.funtester.frame.FunPhaser;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class FunPhaserDemo {
    public static void main(String[] args) throws InterruptedException {
        // Create FunPhaser instance
        FunPhaser phaser = new FunPhaser();
        // Fixed thread pool of size 5
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        // Register and run 10 tasks
        for (int i = 0; i < 10; i++) {
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    phaser.register();
                    try { Thread.sleep(1000); } catch (InterruptedException e) { e.printStackTrace(); }
                    phaser.done();
                }
            });
        }
        // Wait for all tasks to finish
        phaser.await();
        System.out.println("Completed task count: " + phaser.queryTaskNum());
        executorService.shutdown();
    }
}

The demo creates a FunPhaser object, registers ten asynchronous tasks in a pool of five threads, marks each task as done, waits for all tasks, and finally prints the number of completed tasks.

Custom Keyword Integration

/**
 * Execute a closure with FunPhaser synchronization.
 * @param f   Code block to run
 * @param phaser  FunPhaser instance
 */
public static void fun(Closure f, FunPhaser phaser) {
    if (phaser != null) phaser.register();
    ThreadPoolUtil.executeSync(() -> {
        try { f.call(); }
        finally {
            if (phaser != null) {
                phaser.done();
                logger.info("async task {}", phaser.queryTaskNum());
            }
        }
    });
}

This utility method registers a task, runs the provided closure, and ensures the task is marked as done, logging the progress.

Comparison with the Original Phaser Implementation

The original code used java.util.concurrent.Phaser with register() and arrive() methods, which suffered from the party‑count limit. The new FunPhaser replaces Phaser, removes the limit, and simplifies the API by using register() and done(). Overall, the new implementation is more concise and easier to use.