Java 8 Completable Future

Question

My problem is how to use Completable Future.

I have a class that implements Callable.

public class Executor implements Callable<Collection>

Earlier is used to do -

service.submit(collectorService);

Which would return a Future<Collection>. However we don't want to use future anymore and need CompletableFuture . One idea is that we don't need to poll with CompletableFuture and We don't have to wait and block until it's ready.

So how would i use completable future and call a function say isDone() when the callable thread finishes.

Vyner

Maybe interesting to you : #23302098 – Churr

“we don't want to use future anymore and need CompletableFuture”— interesting mix of “want” and “need”. Doesn’t sound like a rational decision to me… – Relegate

@Relegate : agree with u. Should rephrase my words. We had implemented Future<T> but seems the team that would consume our code does not want to do a poll of Future.isDone() and so would like us to supply CompletableFuture<T> – Vyner

Answer 1

Given a CompletableFuture<T> f, you can kick off a synchronous or asynchronous task to run upon completion using:

f.thenApply(result -> isDone(result));      // sync callback
f.thenApplyAsync(result -> isDone(result)); // async callback

...or, if you don't need the result:

f.thenRun(() -> isDone());
f.thenRunAsync(() -> isDone());

Mahratta

Sorry for asking more but how do i submit my Executor class objects inside the thenApply Functions. What does result object refer to here? – Vyner

The Consumer<T> passed to thenAccept[Async] is a callback which operates on the result of the task, so result refers to the result of the future f. In your original example, it would be a Collection. You can reference any final (or "effectively final") reference from your callback lamda, including your executor, if you want to submit another task in your callback. – Mahratta

thenApply is also async, the XXXAsync methods mean execute in a different thread. – Drogin

Answer 2

You can create a lambda expression that calls your existing collectorService. A Supplier lambda expression that CompletableFuture.supplyAsync will accept would look something like this

 Supplier<Collection> supplier = () -> collectorService.call();

And can be used with the CompletableFuture as follows

  CompletableFuture.supplyAsync(() -> collectorService.call(),service)
         .thenApply(collection->isDone(collection);

As others have pointed out the thenApply will be executed when the collectorService.call() method returns a result - on the same thread that performed our Future task. Using thenApplyAsync would resubmit another task to an executor service (raw performance is about an order of magnitude slower so don't do this unless you have a good reason too!).

Mccollough

Answer 3

If I understand you correctly, you want to know how to submit a "task" (your previous "Executor") that gives back a CompletableFuture.

You do this by calling

CompletableFuture.supplyAsync(collectorService)

The difference is that your "Executor" must implement now Supplier instead of Callable

Employ

Yep, the collectorService could also be wrapped inside a Supplier lambda expression if changing the source wasn't an option. e.g. CompletableFuture.supplyAsync(() -> collectorService.call()) – Mccollough

@JohnMcClean I think your comment is the best answer and I think you should make it an actual answer. – Brack

@Brack sure, added a specific answer with some more detail. – Mccollough

Answer 4

We don't pass runnable or callable in completableFuture. It takes supplier type which is a functional interface. Just create normal methods and pass them with the object of executor. For reference conside the below example .

package completableFuture;

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

public class CompFuture {
    ExecutorService firstExecService = Executors.newFixedThreadPool(5);

    public static void main(String[] args) {

        CompFuture compFuture = new CompFuture();
        compFuture.testMe("Java");
    }

    public String m1(String param) {

        Random r = new Random();
        int val = r.nextInt(20) * 1000;
        System.out.println(Thread.currentThread().getName() + " " + val);

        try {
            Thread.sleep(val);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return param + " Hello my";
    }

    public void m2(String salutation) {
        System.out.println(Thread.currentThread().getName() + "  ##" + salutation + " Friend!");
    }

    public void testMe(String start) {
        System.out.println("TM: " + Thread.currentThread());

        for (int i = 0; i < 5; i++) {
            CompletableFuture.supplyAsync(() -> m1(start), firstExecService).thenAccept(s -> m2(s));
        }
    }

}

Output of above program:: Thread which takes min time to execute gives its output first.

TM: Thread[main,5,main]

pool-1-thread-1 1000

pool-1-thread-2 14000

pool-1-thread-4 3000

pool-1-thread-3 0

pool-1-thread-5 9000

pool-1-thread-3 ##Java Hello my Friend!

pool-1-thread-1 ##Java Hello my Friend!

pool-1-thread-4 ##Java Hello my Friend!

pool-1-thread-5 ##Java Hello my Friend!

pool-1-thread-2 ##Java Hello my Friend!

Blondell