认识达内从这里开始

某些业务场景需要在一个异步操作完成后做一个回调，如果使用Future有个问题，需要主动调用get然后再执行逻辑。但是异步逻辑不知道什么时候结束，所以推荐回调的方式处理比较优雅，本文总结了三种方式可以实现此类需求。

1、CompletableFuture

Future可以明确地完成(设定其值和状态)，并且可以被用作CompletionStage ，支持有关的功能和它的完成时触发动作。

当两个或多个线程试图complete ， completeExceptionally ，或cancel一个CompletableFuture，只有一个成功。

除了直接操作状态和结果的这些和相关方法外，CompletableFuture还实现了接口CompletionStage ，具有以下策略：

为异步方法的依赖完成提供的操作可以由完成当前CompletableFuture的线程或完成方法的任何其他调用者执行。

所有不使用显式Executor参数的异步方法都使用ForkJoinPool.commonPool()执行 (除非它不支持至少两个并行级别，在这种情况下，使用新的线程)。 为了简化监视，调试和跟踪，所有生成的异步任务都是标记接口CompletableFuture.AsynchronousCompletionTask的实例 。

所有CompletionStage方法都是独立于其他公共方法实现的，因此一个方法的行为不会受到子类中其他方法的覆盖的影响。

CompletableFuture还实施Future ，具有以下政策：

由于(不同于FutureTask )，这个类不能直接控制导致其完成的计算，所以取消被视为另一种形式的异常完成。 方法cancel具有相同的效果completeExceptionally(new CancellationException()) 。 方法isCompletedExceptionally()可用于确定CompletableFuture是否以任何特殊方式完成。

如果使用CompletionException异常完成，方法get()和get(long, TimeUnit)将抛出与对应的CompletionException中保持的相同原因的ExecutionException。 为了简化大多数情况下的使用，此类还定义了方法join()和getNow(T) ，而是在这些情况下直接抛出CompletionException。

public U CompletionStage U thenApply(Function fn);

public U CompletionStage U thenApplyAsync(Function fn);

public U CompletionStage U thenApplyAsync(Function fn,Executor executor);

2、Guava 的 ListenableFuture

ListenableFuture 是对原有 Future 的增强，可以用于监听 Future 任务的执行状况，是执行成功还是执行失败，并提供响应的接口用于对不同结果的处理。简单来说就是他提供了一个回调接口，在任务结束时可以主动去做通知调用，有点主动推的感觉。

public class ListenFutureTest {

public static void main(String[] args) {

testListenFuture();

}

public static void testListenFuture() {

System.out.println("主任务执行完，开始异步执行副任务1.....");

ListeningExecutorService pool = MoreExecutors.listeningDecorator(Executors.newFixedThreadPool(5));

ListenableFuture future = pool.submit(new Task());

Futures.addCallback(future, new FutureCallback() {

@Override

public void onSuccess(String result) {

System.out.println("成功,结果是:" + result);

}

@Override

public void onFailure(Throwable t) {

System.out.println("出错,业务回滚或补偿");

}

});

System.out.println("副本任务启动,回归主任务线，主业务正常返回2.....");

}

}

class Task implements Callable {

@Override

public String call() throws Exception {

TimeUnit.SECONDS.sleep(1);

// int a =1/0;

return "task done";

}

}

3、RxJava2

推荐使用RxJava2，具有背压和默认线程调度功能。

ExecutorService executor = Executors.newFixedThreadPool(2);

System.out.println("MAIN: " + Thread.currentThread().getId());

Callable callable = () -> {

System.out.println("callable [" + Thread.currentThread().getId() + "]: ");

Path filePath = Paths.get("build.gradle");

return Files.readAllLines(filePath).stream().flatMap(s -> Arrays.stream(s.split

(""))).count() + "";

};

Future future = executor.submit(callable);

Consumer onNext = v -> System.out

.println("consumer[" + Thread.currentThread().getId() + "]:" + v);

Flowable.fromCallable(callable).subscribe(onNext);

Flowable.fromFuture(future).subscribe(onNext);

System.out.println("END");