CompletableFuture优雅处理并发最佳实践

1、 supplyAsync方法需要一个Supplier函数接口，通常用于执行异步计算

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    dosomething("处理事务");
    return "结果";
});

2、 runAsync接受一个Runnable函数接口，不关心异步任务的结果

CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
    dosomething("处理事务");
});

3、 组合使用

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
    simulateTask("task-1");
    return "task-1";
});

CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
    simulateTask("task-2");
    return "task-2";
});

// 组合两个future，等待它们都完成
CompletableFuture<String> combinedFuture = future1.thenCombine(future2, (task-1, task-2) -> {
    return "==== " + task-1+ "，" + task-2;
});

4、链式调用

• thenApply用于处理和转换CompletableFuture的结果
• thenAccept用于消费CompletableFuture的结果，不返回新的CompletableFuture
• thenRun则不关心前一个任务的结果，只是在前一个任务执行完后，执行一些后续操作

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    simulateTask("task-1");
    return "task-1";
});
future.thenApply(result -> {
    return "处理后的结果：" + result;
}).thenAccept(processedResult -> {
    System.out.println("最终结果：" + processedResult);
}).thenRun(() -> {
    System.out.println("所有操作完成");
});

5、 基本异常处理

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    if (new Random().nextBoolean()) {
        throw new RuntimeException("出错啦!");
    }
    return "正常结果";
}).exceptionally(ex -> {
    return "错误：" + ex.getMessage();
});

future.thenAccept(System.out::println);

6、异常继续处理

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    if (new Random().nextBoolean()) {
        throw new RuntimeException("出错啦!");
    }
    return "正常结果";
}).handle((result, ex) -> {
    if (ex != null) {
        return "处理异常：" + ex.getMessage();
    }
    return "处理结果：" + result;
});
future.thenAccept(System.out::println);

7、 依赖关系

CompletableFuture<String> masterFuture = CompletableFuture.supplyAsync(() -> {
    simulateTask("task-1");
    return "task-1";
});

CompletableFuture<String> dependentFuture = masterFuture.thenCompose(result -> {
    return CompletableFuture.supplyAsync(() -> {
        simulateTask("处理依赖于" + result + "的数据");
        return "处理后的数据";
    });
});

dependentFuture.thenAccept(System.out::println);

8、 处理多个Future

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
    simulateTask("任务一");
    return "结果一";
});

CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
    simulateTask("任务二");
    return "结果二";
});

CompletableFuture<Void> allFutures = CompletableFuture.allOf(future1, future2);

allFutures.thenRun(() -> {
    System.out.println("所有任务完成");
});

CompletableFuture默认情况下使用ForkJoinPool，但在某些场景下使用自定义线程池可更好地控制资源

ExecutorService customExecutor = Executors.newFixedThreadPool(10);
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    return "使用自定义线程池";
}, customExecutor);

当组合多个CompletableFuture，对每一个Future都进行错误处理，避免一个未捕获的异常破坏整个操作链

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> "任务1").exceptionally(ex -> "默认值1");
CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> "任务2").exceptionally(ex -> "默认值2");
CompletableFuture<String> combinedFuture = future1.thenCombine(future2, (result1, result2) -> result1 + " 和 " + result2);