ExecutorService: Better Thread-flow Regulation

Question

I have a question based on the ExecutorService Thread flow regulation: I would like to .submit() multiple threads for executon where I would like some threads to wait until specific previous Threads have finished executing. .

So far I know of one such method of doing this by using CountDownLatch(). Following example illustrates 2 threads that need to finish before the third can start:

public class Example{

public static void main(String[] args) throws InterruptedException {

    CountDownLatch cdl = new CountDownLatch(2); //count of 2
    ExecutorService executor = Executors.newCachedThreadPool();
    executor.submit(new Runnable() {
        public void run() {
             method1(); //simulation of a useful method 1
            cdl.countDown(); //reduces the count by one
        }
    });
    executor.submit(new Runnable() {
        public void run() {
             method2(); //simulation of a useful method 2
            cdl.countDown();
        }
    });
    cdl.await(); //will wait until both methods are complete
    executor.submit(new Runnable() { 
        public void run() {
        result(); //simulation of a method that needs previous threads to execute }
        });
    }
}

Quite obvious, such method is not the optimal for such work as, for one thing, one cannot add additional threads after the .await() that do not depend on the CDL itself.

Therefore, what are the better alternatives to regulate Thread flow with ExecutorService that allow better manipulation of Thread execution compared to CDL.

Answer 1

Just wait for CountDownLatch inside 3rd Runnable

executor.submit(new Runnable() { 
    public void run() {
       cdl.await(); //will wait until both methods are complete
       result(); //simulation of a method that needs previous threads to execute }
    });
}

You might need to declare cdl as final to reference it inside anonymous Runnable instance:

final CountDownLatch cdl = new CountDownLatch(2); //count of 2 

Alternative approach

have a method which will create 3rd task like this:

void createThirdTask() {
    executor.submit(new Runnable() { 
        public void run() {
           result(); //simulation of a method that needs previous threads to execute }
        });
    }
}

Have one shared lock between first two tasks and counter.

private ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
private int count = 2;

Inside method1() and method2() decrement value of CDL and fire third task if it reached zero.

void method1() {
        //your actual code goes here
    try {
        lock.writeLock().lock();
        if(count-- == 0) {
            createThirdTask();
        }
    } finally {
        lock.writeLock().unlock()
    }
}

ReentrantReadWriteLock is there to prevent race conditions.

Answer 2

I would suggest to use CompletableFuture for such kind of thread tasks. It might look like this :

public class RunAfterNThreads {

public static void main(String[] args) throws ExecutionException, InterruptedException {
    CompletableFuture.supplyAsync(RunAfterNThreads::runFirstBatchOfThreads)
            .thenAcceptAsync((t) -> runSecondBatchOfThreads(null)).get();
}

private static Object runSecondBatchOfThreads(Object something) {
    return something;
}

private static <U> U runFirstBatchOfThreads() {
    return null;
}

}

If you want to use some library for this, i would suggest to explore something like Akka (event driven).

Answer 3

I am not sure if Java really has a good standard mechanism in place for you to do that. The internal code for Eclipse has quite some extra classes to handle tasks and their dependencies.

However, I think you might be able to "use" CompletableFuture to do so:

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Main {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newCachedThreadPool();
        Runnable r1 = () -> System.out.println("runnable 1");
        Runnable r2 = () -> System.out.println("runnable 2");
        Runnable r3 = () -> System.out.println("runnable 3");


        CompletableFuture<Void> cf1 = CompletableFuture.runAsync(r1, executor);
        CompletableFuture<Void> cf2 = CompletableFuture.runAsync(r2, executor);

        cf1.runAfterBoth(cf2, r3);
  }
}

If your situation is more complex you might be better off searching for a 'Directed Acyclic Graph' task library.