Java - CompletableFuture

CompletableFuture是Java8中对Future的增强。实现CompletionStage接口(40余个方法)，大多数方法多数应用在函数式编程中，并且支持流式调用。

简单实现：

public class CompletableFutureTest implements Runnable {

    CompletableFuture<Integer> completableFuture = null;

    public CompletableFutureTest(CompletableFuture<Integer> completableFuture) {
        this.completableFuture = completableFuture;
    }

    @Override
    public void run() {
        int result = 0;
        try {
            result = completableFuture.get() * completableFuture.get();
        } catch (Exception e) {

        }
        System.out.println(result);
    }


    public static void main(String[] args) throws InterruptedException, ExecutionException {
        final CompletableFuture<Integer> future = new CompletableFuture<>();
        new Thread(new CompletableFutureTest(future)).start();
        //模拟长时间的计算过程
        Thread.sleep(1000);
        //告知完成结果
        future.complete(60);

    }
}

Future最令人诟病的就是要等待，要自己去检查任务是否完成了，在Future中，任务完成的时间是不可控的。而CompletableFuture的最大改进在于，任务完成的时间也开放了出来。

CompletableFuture的异步执行：

public static Integer calc(Integer param) {
	try {
		//模拟长时间的执行
		Thread.sleep(1000);
	} catch (InterruptedException e) {

	}
	return param * param;
}

public static void main(String[] args) throws InterruptedException, ExecutionException {

	//异步调用
	final CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> calc(50));
	System.out.println(future2.get());
}

CompletableFuture的流式调用：

public static Integer calc(Integer param) {
	try {
		//模拟长时间的执行
		Thread.sleep(1000);
	} catch (InterruptedException e) {

	}
	return param * param;
}


public static void main(String[] args) throws InterruptedException, ExecutionException {

	//流式调用
	CompletableFuture<Void> future3 = CompletableFuture.supplyAsync(() -> calc(40))
			.thenApply((i) -> Integer.toString(i))
			.thenApply((str) -> "\"" + str + "\"")
			.thenAccept(System.out::println);
	future3.get();
}

组合多个CompletableFuture：

public static Integer calc2(Integer param) {
	return param / 2;
}

public static void main(String[] args) throws InterruptedException, ExecutionException {
	//组合多个CompletableFuture
	CompletableFuture<Void> future4 = CompletableFuture.supplyAsync(() -> calc2(100))
			.thenCompose((i) -> CompletableFuture.supplyAsync(() -> calc2(i)))
			.thenApply((str) -> "\"" + str + "\"")
			.thenAccept(System.out::println);
	future4.get();
}

CompletableFuture跟性能上关系不大，更多的是为了支持函数式编程，在功能上的增强。当然开放了完成时间的设置是一大亮点。

演示的完整代码如下：

package main.java.com.study.completableFuture;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * @author: whb
 * @description: 增强版Future测试类
 */
public class CompletableFutureTest implements Runnable {

    CompletableFuture<Integer> completableFuture = null;

    public CompletableFutureTest(CompletableFuture<Integer> completableFuture) {
        this.completableFuture = completableFuture;
    }

    @Override
    public void run() {
        int result = 0;
        try {
            result = completableFuture.get() * completableFuture.get();
        } catch (Exception e) {

        }
        System.out.println(result);
    }

    public static Integer calc(Integer param) {
        try {
            //模拟长时间的执行
            Thread.sleep(1000);
        } catch (InterruptedException e) {

        }
        return param * param;
    }

    public static Integer calc2(Integer param) {
        return param / 2;
    }

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        final CompletableFuture<Integer> future = new CompletableFuture<>();
        new Thread(new CompletableFutureTest(future)).start();
        //模拟长时间的计算过程
        Thread.sleep(1000);
        //告知完成结果
        future.complete(60);

        //异步调用
        final CompletableFuture<Integer> future2 = CompletableFuture.supplyAsync(() -> calc(50));
        System.out.println(future2.get());

        //流式调用
        CompletableFuture<Void> future3 = CompletableFuture.supplyAsync(() -> calc(40))
                .thenApply((i) -> Integer.toString(i))
                .thenApply((str) -> "\"" + str + "\"")
                .thenAccept(System.out::println);
        future3.get();

        //组合多个CompletableFuture
        CompletableFuture<Void> future4 = CompletableFuture.supplyAsync(() -> calc2(100))
                .thenCompose((i) -> CompletableFuture.supplyAsync(() -> calc2(i)))
                .thenApply((str) -> "\"" + str + "\"")
                .thenAccept(System.out::println);
        future4.get();
    }
}