Java8 CompletableFuture异步多线程怎么实现

目录

1. 引言
2. CompletableFuture简介
3. 创建CompletableFuture
4. 异步执行任务
5. 任务链式调用
6. 异常处理
7. 组合多个CompletableFuture
8. 超时处理
9. CompletableFuture与线程池
10. 实际应用场景
11. 总结

引言

在现代软件开发中，异步编程和多线程处理变得越来越重要。Java 8引入了CompletableFuture类，它提供了一种强大的方式来处理异步任务和多线程编程。本文将详细介绍如何使用CompletableFuture来实现异步多线程编程，并探讨其在实际应用中的使用场景。

CompletableFuture简介

CompletableFuture是Java 8引入的一个类，它实现了Future接口，并提供了更丰富的功能来处理异步任务。与传统的Future相比，CompletableFuture支持链式调用、异常处理、任务组合等高级功能，使得异步编程更加灵活和强大。

创建CompletableFuture

1. 使用CompletableFuture.runAsync()

runAsync()方法用于执行一个没有返回值的异步任务。

CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
    // 异步任务代码
    System.out.println("异步任务执行中...");
});

2. 使用CompletableFuture.supplyAsync()

supplyAsync()方法用于执行一个有返回值的异步任务。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 异步任务代码
    return "异步任务执行结果";
});

异步执行任务

1. 使用thenApply()

thenApply()方法用于在异步任务完成后对结果进行处理。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    return "Hello";
}).thenApply(result -> {
    return result + " World";
});

System.out.println(future.get()); // 输出: Hello World

2. 使用thenAccept()

thenAccept()方法用于在异步任务完成后消费结果。

CompletableFuture.supplyAsync(() -> {
    return "Hello";
}).thenAccept(result -> {
    System.out.println(result + " World");
});

3. 使用thenRun()

thenRun()方法用于在异步任务完成后执行一个没有返回值的操作。

CompletableFuture.supplyAsync(() -> {
    return "Hello";
}).thenRun(() -> {
    System.out.println("任务完成");
});

任务链式调用

CompletableFuture支持链式调用，可以将多个异步任务串联起来执行。

CompletableFuture.supplyAsync(() -> {
    return "Hello";
}).thenApply(result -> {
    return result + " World";
}).thenApply(result -> {
    return result + "!";
}).thenAccept(result -> {
    System.out.println(result); // 输出: Hello World!
});

异常处理

1. 使用exceptionally()

exceptionally()方法用于处理异步任务执行过程中发生的异常。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException("任务执行失败");
}).exceptionally(ex -> {
    System.out.println("捕获异常: " + ex.getMessage());
    return "默认值";
});

System.out.println(future.get()); // 输出: 默认值

2. 使用handle()

handle()方法用于在异步任务完成后处理结果或异常。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    throw new RuntimeException("任务执行失败");
}).handle((result, ex) -> {
    if (ex != null) {
        System.out.println("捕获异常: " + ex.getMessage());
        return "默认值";
    }
    return result;
});

System.out.println(future.get()); // 输出: 默认值

组合多个CompletableFuture

1. 使用thenCompose()

thenCompose()方法用于将一个CompletableFuture的结果作为另一个CompletableFuture的输入。

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

CompletableFuture<String> future2 = future1.thenCompose(result -> {
    return CompletableFuture.supplyAsync(() -> {
        return result + " World";
    });
});

System.out.println(future2.get()); // 输出: Hello World

2. 使用thenCombine()

thenCombine()方法用于将两个CompletableFuture的结果进行组合。

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

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

CompletableFuture<String> combinedFuture = future1.thenCombine(future2, (result1, result2) -> {
    return result1 + " " + result2;
});

System.out.println(combinedFuture.get()); // 输出: Hello World

3. 使用allOf()

allOf()方法用于等待多个CompletableFuture全部完成。

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

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

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

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

4. 使用anyOf()

anyOf()方法用于等待多个CompletableFuture中的任意一个完成。

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

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

CompletableFuture<Object> anyFuture = CompletableFuture.anyOf(future1, future2);

System.out.println(anyFuture.get()); // 输出: Hello 或 World

超时处理

CompletableFuture本身不直接支持超时处理，但可以通过CompletableFuture与ScheduledExecutorService结合来实现。

ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    try {
        Thread.sleep(5000); // 模拟长时间任务
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    return "任务完成";
});

CompletableFuture<String> timeoutFuture = new CompletableFuture<>();
scheduler.schedule(() -> {
    timeoutFuture.completeExceptionally(new TimeoutException("任务超时"));
}, 2, TimeUnit.SECONDS);

CompletableFuture<String> resultFuture = future.applyToEither(timeoutFuture, Function.identity());

try {
    System.out.println(resultFuture.get()); // 输出: 任务超时
} catch (ExecutionException e) {
    System.out.println(e.getCause().getMessage()); // 输出: 任务超时
}

CompletableFuture与线程池

CompletableFuture默认使用ForkJoinPool.commonPool()来执行异步任务，但也可以通过指定自定义的线程池来控制任务的执行。

ExecutorService customExecutor = Executors.newFixedThreadPool(10);

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    return "Hello";
}, customExecutor);

System.out.println(future.get()); // 输出: Hello

实际应用场景

1. 并行处理多个任务

CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
    return "任务1完成";
});

CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
    return "任务2完成";
});

CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
    return "任务3完成";
});

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

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

2. 异步调用外部服务

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // 调用外部服务
    return callExternalService();
}).thenApply(result -> {
    // 处理外部服务返回的结果
    return processResult(result);
}).exceptionally(ex -> {
    // 处理异常
    return handleException(ex);
});

3. 批量处理数据

List<CompletableFuture<String>> futures = new ArrayList<>();

for (int i = 0; i < 10; i++) {
    CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
        return processData(i);
    });
    futures.add(future);
}

CompletableFuture<Void> allFutures = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));

allFutures.thenRun(() -> {
    System.out.println("所有数据处理完成");
});

总结

CompletableFuture是Java 8中引入的一个强大的工具，它提供了丰富的功能来处理异步任务和多线程编程。通过CompletableFuture，我们可以轻松地实现任务的链式调用、异常处理、任务组合等高级功能。在实际应用中，CompletableFuture可以用于并行处理多个任务、异步调用外部服务、批量处理数据等场景。掌握CompletableFuture的使用，将极大地提高我们的异步编程能力和代码的可维护性。

---

以上是关于Java 8中CompletableFuture异步多线程实现的详细介绍。希望本文能帮助你更好地理解和使用CompletableFuture，并在实际项目中发挥其强大的功能。