Asynchronous Programming

[charoi]

Topic:

Asynchronus Programming

Hi! Just wanted to check the Java API, what is the difference between thenComposeAsyncand thenCompose ? Also, for async programming, will CompletionStage<T> ever be used? It seems like the class only ever returns CompletionStage objects.

[bryanwee023]

I was playing with CompletableFuture in jshell, seems like regular callbacks runs on the main thread if the initial asynchronised task is complete, but on the side thread if it is incomplete (at time of assigning callback). But the async variants run on the side thread regardless. Is that the main diff?

[bentanjunrong]

From my understanding, it looks like the difference is minimal.

CompletableFuture.supplyAsync(() -> getA())
    .thenApply(a -> foo(a))
    .thenApply(b -> bar(b));

For thenApply(), the thread that it gets called on is the same thread as the upstream task. In this example, the thread that runs Line 2/3 is the same as the thread that runs supplyAsync(() -> getA()).

If we instead use thenApplyAsync():

CompletableFuture.supplyAsync(() -> getA())
    .thenApplyAsync(a -> foo(a))
    .thenApplyAsync(b -> bar(b));

then Lines 2/3 are run on threads from the default ForkJoinPool.commonPool(). However, the documentation seems vague, and it could be possible that they end up running on the same thread as anyway (see https://stackoverflow.com/questions/46060438/in-which-thread-does-completablefutures-completion-handlers-execute-in).

tl;dr I don't see any practical difference, so for our uses its probably safe to just stick to thenApply().

[bentanjunrong]

Okay I just took a look at the documentation for thenApplyAsync(), and you can see that it is an overloaded method.

thenApplyAsync(Function<? super T,? extends U> fn, Executor executor) Returns a new CompletionStage that, when this stage completes normally, is executed using the supplied executor, with this stage's result as the argument to the supplied function.

So if you don't want it to run on ForkJoinPool.commonPool(), you can specify which thread pool you want it to run from.

ExecutorService pool_1 = Executors.newFixedThreadPool(10);
ExecutorService pool_2 = Executors.newFixedThreadPool(50);
ExecutorService pool_3 = Executors.newFixedThreadPool(1000);

CompletableFuture.supplyAsync(() -> getA(), pool_1)
    .thenApplyAsync(a -> foo(a), pool_2)
    .thenApplyAsync(b -> bar(b), pool_3);

thenApplyAsync() could have practical use for larger scale applications where thread management is important, e.g. a -> foo(a) is a costly transformation. So in our case, if we want to specify the thread pool for individual function calls, then we have to use thenApplyAsync(), otherwise thenApply() should suffice.

[bryanwee023]

@bentanjunrong I think thenApplyAsync() without the executor parameter guarantees that your callback runs on the non-main thread. So it might be useful if your callback is something time expensive and can be run as an asynchronised task.

thenApply() could run on the main or forked thread depending on when it's called. So it's meant for more trivial tasks.