Asynchronous artifact completable future

background

CompletableFuture, There are a lot of functional interfaces , When the interface reaches the bottleneck , It is likely that you need to use multithreading to optimize your project

Want to facilitate asynchronous execution of tasks , It has to be put into a separate thread . Inherit Thread class , Realization Runnable Can't get the results of the mission , At this point, we have to mention another way to create threads , Realization Callable Interface .

For simple scenarios use Future There's nothing inconvenient . But some complex scenes are troublesome ,
Such as 2 Asynchronous tasks , If one of them has a result, it goes back to .Future It's not convenient to use , Because when you want to get results , Or execute future.get() Method , But this blocks the thread , It becomes synchronous operation , Or polling isDone() Method , But it's more expensive CPU resources .

Netty and Google guava To solve this problem , stay Future On the basis of this, the observer model is introduced （ That is to say Future On addListener）, Notify the listener when the result of the calculation is complete .

Usage mode

// Asynchronous thread 
 Threads 1 -> do somthing -> then do somthing
// Asynchronous thread 
 Threads 2 -> do somthing -> then do somthing

Provides a more comprehensive approach , And there are ways to handle exceptions , It is convenient to use

Java8 Newly added CompletableFuture It draws Netty Wait right Future The reconstruction of , Simplify the complexity of asynchronous programming , And it provides the ability of functional programming

establish CompletableFuture object

Method name	describe
completedFuture(U value)	Return a calculated CompletableFuture
runAsync(Runnable runnable)	Use ForkJoinPool.commonPool() Perform tasks as a thread pool , no return value
runAsync(Runnable runnable, Executor executor)	Use the specified thread pool to perform tasks , no return value
supplyAsync(Supplier supplier)	Use ForkJoinPool.commonPool() Perform tasks as a thread pool , There is a return value
supplyAsync(Supplier supplier, Executor executor)	Use the specified thread pool to perform tasks , There is a return value

@FunctionalInterface
public interface Supplier<T> {
    
    T get();
}

Supplier It is a functional interface that can get the return value

CompletableFuture<Integer> intFuture = CompletableFuture.completedFuture(100);
// 100
System.out.println(intFuture.get());

CompletableFuture<Void> voidFuture = CompletableFuture.runAsync(() -> System.out.println("hello"));
// null
System.out.println(voidFuture.get());

CompletableFuture<String> stringFuture = CompletableFuture.supplyAsync(() -> "hello");
// hello
System.out.println(stringFuture.get());

At the end of the calculation

Method name
whenComplete(BiConsumer<? super T,? super Throwable> action)
whenCompleteAsync(BiConsumer<? super T,? super Throwable> action)
whenCompleteAsync(BiConsumer<? super T,? super Throwable> action, Executor executor)

Because it is BiConsumer<? super T,? super Throwable> Functional interface , So you can handle normal and abnormal calculations

whenComplete and whenCompleteAsync The difference is as follows

1. whenComplete： The thread that finished executing the current task continues to execute whenComplete The task of
2. whenCompleteAsync： hold whenCompleteAsync This task is submitted to the thread pool for execution

CompletableFuture The definition of all methods and whenComplete It's all very similar

1. The method is different Async The end means executing with the same thread
2. Methods to Async The end means that the task is submitted to the thread pool for execution
3. Methods to Async You can end with ForkJoinPool.commonPool() As a thread pool , You can also use your own thread pool

All of the methods described in the following section are written in only one way case

CompletableFuture future = CompletableFuture.supplyAsync(() -> {
    
    return "hello";
}).whenComplete((v, e) -> {
    
    // hello
    System.out.println(v);
});
// hello
System.out.println(future.get());

transformation , consumption , perform

Method name	describe
thenApply	Get the return of the previous task , And return the value of the current task
thenAccept	Get the return of the previous task , Pure consumption , no return value
thenRun	After the last task is completed , Start execution thenRun The task

CompletableFuture.supplyAsync(() -> {
    
    return "hello ";
}).thenAccept(str -> {
    
    // hello world
    System.out.println(str + "world");
}).thenRun(() -> {
    
    // task finish
    System.out.println("task finish");
});

Combine （ Both tasks are completed ）

Method name	describe
thenCombine	Combine two future, Get two future Return result of , And return the return value of the current task
thenAcceptBoth	Combine two future, Get two future The return result of the task , And then deal with the task , no return value
runAfterBoth	Combine two future, There is no need to get future Result , Just two future After finishing the task , Deal with the task

CompletableFuture future = CompletableFuture.supplyAsync(() -> {
    
    return "today ";
}).thenApply(t -> {
    
    return t + "is ";
}).thenCombine(CompletableFuture.completedFuture("tuesday"), (t, u) -> {
    
    return t + u;
}).whenComplete((t, e) -> {
    
    // today is tuesday
    System.out.println(t);
});

Combine （ Just one task to complete ）

Method name	describe
applyToEither	One of the two tasks is completed , Get its return value , Process the task and return the return value of the current task
acceptEither	One of the two tasks is completed , Get its return value , Processing tasks , no return value
runAfterEither	One of the two tasks is completed , There is no need to get future Result , Processing tasks , There is no return value

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
    
    sleepRandom();
    return "today is";
});
CompletableFuture future2 = CompletableFuture.supplyAsync(() -> {
    
    sleepRandom();
    return "tuesday";
});
CompletableFuture future = future1.applyToEither(future2, str -> str);
// today is tuesday  Random output 
System.out.println(future.get());

sleepRandom() A random pause function written for me

Multitasking

Method name	describe
allOf	When all CompletableFuture Perform the calculation when finished
anyOf	Any one CompletableFuture Perform the calculation when finished

allOf Use

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
    
    sleepRandom();
    return "today";
});
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
    
    sleepRandom();
    return "is";
});
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
    
    sleepRandom();
    return "tuesday";
});
// today is tuesday
CompletableFuture.allOf(future1, future2, future3)
        .thenApply(v ->
                Stream.of(future1, future2, future3)
                        .map(CompletableFuture::join)
                        .collect(Collectors.joining(" ")))
        .thenAccept(System.out::print);

anyOf Use

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
    
    sleepRandom();
    return "today";
});
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
    
    sleepRandom();
    return "is";
});
CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
    
    sleepRandom();
    return "tuesday";
});
CompletableFuture<Object> resultFuture = CompletableFuture.anyOf(future1, future2, future3);
// today is tuesday  Random output 
System.out.println(resultFuture.get());

exception handling

Method name	describe
exceptionally	Capture exception , To deal with

CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
    
    return 100 / 0;
}).thenApply(num -> {
    
    return num + 10;
}).exceptionally(throwable -> {
    
    return 0;
});
// 0
System.out.println(future.get());

Of course, there are interfaces that catch exceptions

CompletableFuture future = CompletableFuture.supplyAsync(() -> {
    
    String str = null;
    return str.length();
}).whenComplete((v, e) -> {
    
    if (e == null) {
    
        System.out.println(" The normal result is " + v);
    } else {
    
        //  Something is wrong java.util.concurrent.CompletionException: java.lang.NullPointerException
        System.out.println(" Something is wrong " + e.toString());
    }
});

We can also use exceptions to identify , Throw an exception in the main thread

// Asynchronous execution result ID 
AtomicBoolean flag = new AtomicBoolean(true);

CompletableFuture completableFuture = CompletableFuture.supplyAsync(() -> {
    
  //do somthing
	return xxx;
}).exceptionally(throwable -> {
    
  flag.set(false);
  return null;
});


CompletableFuture completableFutureV2 = CompletableFuture.supplyAsync(() -> {
    
  //do somthing
	return xxx;
}).exceptionally(throwable -> {
    
  flag.set(false);
  return null;
});

Object obj = completableFuture.get();
Object objV2 = completableFutureV2.get();

if (flag.get() == false){
    
  throw new RuntimeException("xxx abnormal ");
}