前言
当初大部分的CPU都是多核，咱们都晓得想要晋升咱们应用程序的运行效率，就必须得充分利用多核CPU的计算能力；Java早曾经为咱们提供了多线程的API，然而实现形式稍微麻烦，明天咱们就来看看Java8在这方面提供的改善。
假如场景
当初你须要为在线教育平台提供一个查问用户详情的API，该接口须要返回用户的根本信息，标签信息，这两个信息寄存在不同地位，须要近程调用来获取这两个信息；为了模仿近程调用，咱们须要在代码外面提早 1s;
public interface RemoteLoader { String load(); default void delay() { try { Thread.sleep(1000L); } catch (InterruptedException e) { e.printStackTrace(); } }}public class CustomerInfoService implements RemoteLoader { public String load() { this.delay(); return "根本信息"; }}public class LearnRecordService implements RemoteLoader { public String load() { this.delay(); return "学习信息"; }}同步形式实现版本
如果咱们采纳同步的形式来实现这个API接口，咱们的实现代码：
@Testpublic void testSync() { long start = System.currentTimeMillis(); List<RemoteLoader> remoteLoaders = Arrays.asList(new CustomerInfoService(), new LearnRecordService()); List<String> customerDetail = remoteLoaders.stream().map(RemoteLoader::load).collect(toList()); System.out.println(customerDetail); long end = System.currentTimeMillis(); System.out.println("总共破费工夫:" + (end - start));}不出所料，因为调用的两个接口都是提早了 1s ，所以后果大于2秒
Future实现的版本
接下来咱们把这个例子用Java7提供的Future来实现异步的版本，看下成果如何呢？代码如下：
@Testpublic void testFuture() { long start = System.currentTimeMillis(); ExecutorService executorService = Executors.newFixedThreadPool(2); List<RemoteLoader> remoteLoaders = Arrays.asList(new CustomerInfoService(), new LearnRecordService()); List<Future<String>> futures = remoteLoaders.stream() .map(remoteLoader -> executorService.submit(remoteLoader::load)) .collect(toList()); List<String> customerDetail = futures.stream() .map(future -> { try { return future.get(); } catch (InterruptedException | ExecutionException e) { e.printStackTrace(); } return null; }) .filter(Objects::nonNull) .collect(toList()); System.out.println(customerDetail); long end = System.currentTimeMillis(); System.out.println("总共破费工夫:" + (end - start));}这次咱们采纳多线程的形式来革新了咱们这个例子，后果还是比较满意的，工夫大略破费了1s多一点
留神：这里我分成了两个Stream，如何合在一起用同一个Stream，那么在用future.get()的时候会导致阻塞，相当于提交一个工作执行完后才提交下一个工作，这样达不到异步的成果这里咱们能够看到尽管Future达到了咱们预期的成果，然而如果须要实现将两个异步的后果进行合并解决就略微麻一些，这里就不细说，前面次要看下CompletableFuture在这方面的改良
Java8并行流
以上咱们用的是Java8之前提供的办法来实现，接下来咱们来看下Java8中提供的并行流来实习咱们这个例子成果怎么呢？
@Testpublic void testParallelStream() { long start = System.currentTimeMillis(); List<RemoteLoader> remoteLoaders = Arrays.asList(new CustomerInfoService(), new LearnRecordService()); List<String> customerDetail = remoteLoaders.parallelStream().map(RemoteLoader::load).collect(toList()); System.out.println(customerDetail); long end = System.currentTimeMillis(); System.out.println("总共破费工夫:" + (end - start));}运行的后果还是相当的称心，破费工夫 1s 多点
和Java8之前的实现比照，咱们发现整个代码会更加的简洁；
接下来咱们把咱们的例子扭转一下，查问用户详情的接口还须要返回视频观看记录，用户的标签信息，购买订单
public class WatchRecordService implements RemoteLoader { @Override public String load() { this.delay(); return "观看记录"; }}public class OrderService implements RemoteLoader { @Override public String load() { this.delay(); return "订单信息"; }}public class LabelService implements RemoteLoader { @Override public String load() { this.delay(); return "标签信息"; }}咱们持续应用Java8提供的并行流来实现，看下运行的后果是否现实
@Testpublic void testParallelStream2() { long start = System.currentTimeMillis(); List<RemoteLoader> remoteLoaders = Arrays.asList( new CustomerInfoService(), new LearnRecordService(), new LabelService(), new OrderService(), new WatchRecordService()); List<String> customerDetail = remoteLoaders.parallelStream().map(RemoteLoader::load).collect(toList()); System.out.println(customerDetail); long end = System.currentTimeMillis(); System.out.println("总共破费工夫:" + (end - start));}然而这次运行的后果不是太现实，破费工夫超过了2秒
CompletableFuture
根本的用法
@Testpublic void testCompletableFuture() { CompletableFuture<String> future = new CompletableFuture<>(); new Thread(() -> { doSomething(); future.complete("Finish"); //工作执行实现后 设置返回的后果 }).start(); System.out.println(future.join()); //获取工作线程返回的后果}private void doSomething() { System.out.println("doSomething...");}这种用法还有个问题，就是工作呈现了异样，主线程会无感知，工作线程不会把异样给抛出来；这会导致主线程会始终期待，通常咱们也须要晓得呈现了什么异样，做出对应的响应；改良的形式是在工作中try-catch所有的异样，而后调用future.completeExceptionally(e) ，代码如下：
@Testpublic void testCompletableFuture() throws ExecutionException, InterruptedException { CompletableFuture<String> future = new CompletableFuture<>(); new Thread(() -> { try { doSomething(); future.complete("Finish"); } catch (Exception e) { future.completeExceptionally(e); } }).start(); System.out.println(future.get());}private void doSomething() { System.out.println("doSomething..."); throw new RuntimeException("Test Exception");}从当初来看CompletableFuture的应用过程须要解决的事件很多，不太简洁，你会感觉看起来很麻烦；然而这只是表象，Java8其实对这个过程进行了封装，提供了很多简洁的操作形式；接下来咱们看下如何革新下面的代码
@Testpublic void testCompletableFuture2() throws ExecutionException, InterruptedException { CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> { doSomething(); return "Finish"; }); System.out.println(future.get());}这里咱们采纳了supplyAsync，这下看起来简洁了许多，世界都亮堂了; Java8不仅提供容许工作返回后果的supplyAsync，还提供了没有返回值的runAsync；让咱们能够更加的关注业务的开发，不须要解决异样谬误的治理
CompletableFuture异样解决
如果说主线程须要关怀工作到底产生了什么异样，须要对其作出相应操作，这个时候就须要用到exceptionally
@Testpublic void testCompletableFuture2() throws ExecutionException, InterruptedException { CompletableFuture<String> future = CompletableFuture .supplyAsync(() -> { doSomething(); return "Finish"; }) .exceptionally(throwable -> "Throwable exception message:" + throwable.getMessage()); System.out.println(future.get());}应用CompletableFuture来实现咱们查问用户详情的API接口
@Testpublic void testCompletableFuture3() throws ExecutionException, InterruptedException { long start = System.currentTimeMillis(); List<RemoteLoader> remoteLoaders = Arrays.asList( new CustomerInfoService(), new LearnRecordService(), new LabelService(), new OrderService(), new WatchRecordService()); List<CompletableFuture<String>> completableFutures = remoteLoaders .stream() .map(loader -> CompletableFuture.supplyAsync(loader::load)) .collect(toList()); List<String> customerDetail = completableFutures .stream() .map(CompletableFuture::join) .collect(toList()); System.out.println(customerDetail); long end = System.currentTimeMillis(); System.out.println("总共破费工夫:" + (end - start));}这里仍然是采纳的两个Stream来实现的，执行的后果如下：
这个后果不太称心，和并行流的后果差不多，耗费工夫 2秒多点；在这种场景下咱们用CompletableFuture做了这么多工作，然而成果不现实，难道就有没有其余的形式能够让它在快一点吗？
为了解决这个问题，咱们必须深刻理解下并行流和CompletableFuture的实现原理，它们底层应用的线程池的大小都是CPU的核数Runtime.getRuntime().availableProcessors()；那么咱们来尝试一下批改线程池的大小，看看成果如何？
自定义线程池，优化CompletableFuture
应用并行流无奈自定义线程池，然而CompletableFuture能够
@Testpublic void testCompletableFuture4() throws ExecutionException, InterruptedException { long start = System.currentTimeMillis(); List<RemoteLoader> remoteLoaders = Arrays.asList( new CustomerInfoService(), new LearnRecordService(), new LabelService(), new OrderService(), new WatchRecordService()); ExecutorService executorService = Executors.newFixedThreadPool(Math.min(remoteLoaders.size(), 50)); List<CompletableFuture<String>> completableFutures = remoteLoaders .stream() .map(loader -> CompletableFuture.supplyAsync(loader::load, executorService)) .collect(toList()); List<String> customerDetail = completableFutures .stream() .map(CompletableFuture::join) .collect(toList()); System.out.println(customerDetail); long end = System.currentTimeMillis(); System.out.println("总共破费工夫:" + (end - start));}咱们应用自定义线程池，设置最大的线程池数量50，来看下执行的后果
这下执行的后果比较满意了，1秒多点；实践上来说这个后果能够始终继续，直到达到线程池的大小50
并行流和CompletableFuture两者该如何抉择
这两者如何抉择次要看工作类型，倡议
- 如果你的工作是计算密集型的，并且没有I/O操作的话，那么举荐你抉择Stream的并行流，实现简略并行效率也是最高的
- 如果你的工作是有频繁的I/O或者网络连接等操作，那么举荐应用
CompletableFuture，采纳自定义线程池的形式，依据服务器的状况设置线程池的大小，尽可能的让CPU繁忙起来
CompletableFuture的其余罕用办法
- thenApply、thenApplyAsync: 如果工作执行实现后，还须要后续的操作，比方返回后果的解析等等；能够通过这两个办法来实现
- thenCompose、thenComposeAsync: 容许你对两个异步操作进行流水线的操作，当第一个操作实现后，将其后果传入到第二个操作中
- thenCombine、thenCombineAsync：容许你把两个异步的操作整合；比方把第一个和第二个操作返回的后果做字符串的连贯操作
总结
- Java8并行流的应用形式
- CompletableFuture的应用形式、异样解决机制，让咱们有机会治理工作执行中发送的异样
- Java8并行流和
CompletableFuture两者该如何抉择 CompletableFuture的罕用办法
原创不易 转载请注明出处：https://silently9527.cn/archi...