CompletableFuture

Running tasks

supplyAsync()

ExecutorService.submit()

CompletableFuture

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { try (InputStream is = new URL("http://www.nurkiewicz.com").openStream()) { log.info("Downloading"); return IOUtils.toString(is, StandardCharsets.UTF_8); } catch (IOException e) { throw new RuntimeException(e); } });

supplyAsync()

ForkJoinPool.commonPool()

CompletableFuture

Executor

ExecutorService pool = Executors.newFixedThreadPool(10); final CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { //... }, pool);

Callbacks and transformations

CompletableFuture

String

CompletableFuture<Integer> intFuture = future.thenApply(s -> s.length());

s.length()

thenApply

String

String

s.length()

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { sleepSeconds(2); return "ABC"; }, pool); future.thenApply(s -> { log.info("First transformation"); return s.length(); }); future.get(); pool.shutdownNow(); pool.awaitTermination(1, TimeUnit.MINUTES); future.thenApply(s -> { log.info("Second transformation"); return s.length(); });

thenApply()

future

CompletableFuture

thenApply()

main

thenApply()

CompletableFuture

Controlling callback's thread pool

*Async

future.thenApplyAsync(s -> { log.info("Second transformation"); return s.length(); });

ForkJoinPool.commonPool()

pool-1-thread-1 | First transformation ForkJoinPool.commonPool-worker-1 | Second transformation

commonPool

future.thenApplyAsync(s -> { log.info("Second transformation"); return s.length(); }, pool2);

pool-1

pool-2

pool-1-thread-1 | First transformation pool-2-thread-1 | Second transformation

Treating callback like another computation step

CompletableFuture

CompletableFuture

//Imagine this is slow and costly CompletableFuture<Integer> strLen(String s) { return CompletableFuture.supplyAsync( () -> s.length(), pool2); } //... CompletableFuture<Integer> intFuture = future.thenCompose(s -> strLen(s));

String

CompletableFuture<Integer>

thenApply()

thenCompose()

CompletableFuture<CompletableFuture<Integer>>

CompletableFuture

applyToEither()

Future

CompletableFuture<CompletableFuture<Integer>> poor = future1.applyToEither(future2, s -> strLen(s));

flatten

flatMap(identity)

flatMap(x -> x)

flatMap()

thenCompose

CompletableFuture<Integer> good = poor.thenCompose(x -> x);

CompletableFuture