百度360必应搜狗淘宝本站头条
mysql distinctjenkins官网mysql 子查询vba splitpython list sort
当前位置：网站首页 > IT技术 > 正文

带你玩转CompletableFuture异步编程

wptr33 2025-02-19 14:10 37 浏览

为什么使用CompletableFuture

一些业务场景我们需要使用多线程异步执行任务，加快任务执行速度。 JDK5新增了Future接口，用于描述一个异步计算的结果。虽然 Future 以及相关使用方法提供了异步执行任务的能力，但是对于结果的获取却是很不方便，我们必须使用Future.get的方式阻塞调用线程，或者使用轮询方式判断 Future.isDone 任务是否结束，再获取结果。这两种处理方式都不是很优雅，相关代码如下:

    @Test
    public void testFuture() {
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        Future future = executorService.submit(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "hello";
        });
        try {
            System.out.println(future.get());
            System.out.println("end");
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
复制代码

此外就是无法解决多个异步任务需要相互依赖的场景，简单点说就是，主线程需要等待子线程任务执行完毕之后在进行执行，这个时候你可能想到了CountDownLatch，没错确实可以解决，代码如下，但是Java8以后我不在认为这是一种优雅的解决方式，接下来我们来了解下CompletableFuture的使用。

    @Test
    public void testCountDownLatch() {
        ExecutorService executorService = Executors.newFixedThreadPool(5);
        CountDownLatch downLatch = new CountDownLatch(2);
        Future orderFuture = executorService.submit(() -> {
            OrderService orderService = new OrderServiceImpl();
            String result = orderService.queryOrderInfo();
            downLatch.countDown();
            return result;
        });

        Future trailFuture = executorService.submit(() -> {
            TrailService trailService = new TrailServiceImpl();
            String result = trailService.queryTrail();
            downLatch.countDown();
            return result;
        });

        try {
            downLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        try {
            System.out.println(orderFuture.get() + trailFuture.get());
            System.out.println("end");
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
复制代码

使用介绍

创建实例

关于CompletableFuture的创建提供了5种方式，第一个就是创建一个具有默认结果的 CompletableFuture，不经常使用，我们常用就是runAsync和supplyAsync，重点关注下这两个静态方法。runAsync是不具有返回值，supplyAsync具有返回值，关于这两个方法都提供一个两种创建形式，一种是默认的使用公共的 ForkJoinPool 线程池执行，这个线程池默认线程数是 CPU 的核数。

另外一种使用就是提供一个主动创建线程池，这样做相比于ForkJoinPool的好处是，可以通过自定义不同场景的线程池，来进行业务划分方便发现问题，还有一个好处就是对于ForkJoinPool这种共享线程来说，一旦阻塞，会照成其他线程无法获取执行机会。

CompletionStage

关于CompletableFuture核心能力就是通过继承Future和CompletionStage来实现的，关于Future就是提供一些异步的能力，如果单存就这样CompletableFuture也就不会那么强大，所以我们核心就是介绍CompletionStage内部常用一些方法。

关于CompletionStage的方法，可以分为两种类型的接口，其中核心方法都提供异步的方式，这里关于异步的方法不进行介绍，基本上原理类似，都是新提供一个线程池去实现任务。

异步回调

关于异步回调可以分为两类，一种是有参数的返回，另外一种是无参数的返回，有参数返回的包括thenApply和thenCompose，无参数返回的包括thenRun和thenAccept。

有参数返回

thenApply 和 thenCompose表示某个任务执行完成后执行的动作，即回调方法，会将该任务的执行结果即方法返回值作为入参传递到回调方法中，也可以理解为串行化的，唯一不同的是thenCompose需要返回一个新的 CompletionStage，整体的使用如下:

    @Test
    public void testCompletableFuture() {
        long start = System.currentTimeMillis();
        CompletableFuture first = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "aaaaaaaaaaaaaaaaaaaaa";
        }).thenApply(x -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return x + "bbbbbbbbbbbbbbbbbbbbbbbb";
        }).thenCompose(x -> CompletableFuture.supplyAsync(x::toUpperCase));
        System.out.println(first.join());
        long end = System.currentTimeMillis();
        System.out.println("耗时" + (end - start) / 1000 + "");
    }
复制代码

无参数返回

thenAccep也是消费上一个任务的动作，将该任务的执行结果即方法返回值作为入参传递到回调方法中，只是无返回值，thenAccep与thenRun方法不同就是没有入参也没有返回值。

    @Test
    public void testCompletableFuture() {
        long start = System.currentTimeMillis();
        CompletableFuture first = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "a";
        }).thenAccept(x -> {
            System.out.println(x);
        });
        first.join();
        long end = System.currentTimeMillis();
        System.out.println("耗时" + (end - start) / 1000 + "秒");
    }

复制代码

异常

CompletableFuture方法执行过程若产生异常，只有get或者join方法的才能获取到异常，针对这种情况CompletableFuture提供三种处理异常的方式。

exceptionally

exceptionally的使用方式类似于 try catch中的catch代码块中异常处理。exceptionally当某个任务执行异常时执行的回调方法，将抛出的异常作为参数传递到回调方法中，如果该任务正常执行，exceptionally方法返回的CompletionStage的result就是该任务正常执行的结果。

    @Test
    public void testCompletableFuture() {
        long start = System.currentTimeMillis();
        CompletableFuture first = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            throw new RuntimeException("test");
        });
        CompletableFuture two = first.exceptionally((x) -> {
            x.printStackTrace();
            return "123";
        });
        two.join();
        long end = System.currentTimeMillis();
        System.out.println("耗时" + (end - start) / 1000 + "秒");
    }
复制代码

whenComplete

whenComplete的使用类似于 try..catch..finanlly 中 finally 代码块，无论是否发生异常，都将会执行的。whenComplete当某个任务执行完成后执行的回调方法，会将执行结果或者执行期间抛出的异常传递给回调方法，如果是正常执行则异常为null，回调方法对应的CompletableFuture的result和该任务一致，如果该任务正常执行，则get方法返回执行结果，如果是执行异常，则get方法抛出异常。

    @Test
    public void testCompletableFuture() {
        long start = System.currentTimeMillis();
        CompletableFuture first = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "aa";
        }).whenComplete((x, throwable) -> {
            // 如果异常存在,打印异常，并且返回默认值
            if (throwable != null) {
                throwable.printStackTrace();
                System.out.println("失败");
            } else {
                System.out.println("成功");
            }
        });
        System.out.println(first.join());
        long end = System.currentTimeMillis();
        System.out.println("耗时" + (end - start) / 1000 + "秒");
    }
复制代码

handle

跟whenComplete基本一致，区别在于handle的回调方法有返回值，且handle方法返回的CompletableFuture的result是回调方法的执行结果或者回调方法执行期间抛出的异常，与原始CompletableFuture的result无关。

    @Test
    public void testCompletableFuture() {
        long start = System.currentTimeMillis();
        CompletableFuture first = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            throw new RuntimeException("test");
        }).handle((x, throwable) -> {
            // 如果异常存在,打印异常，并且返回默认值
            if (throwable != null) {
                throwable.printStackTrace();
                return "异常";
            } else {
                
                return x + "aa";
            }
        });
        System.out.println(first.join());
        long end = System.currentTimeMillis();
        System.out.println("耗时" + (end - start) / 1000 + "秒");
    }
复制代码

组合关系

组合关系关系分为两种，一种是和的关系，一种是或的关系，和关系就是当所有的任务执行完成以后再继续执行，类似于CountDownLatch，或关系就是只要有一个任务执行完成以后就可以向下执行。

和关系

thenCombine /thenAcceptBoth / runAfterBoth/allOf

这四个方法可以将多个CompletableFuture组合起来，将多个CompletableFuture都执行完成以后，才能执行后面的操作，区别在于，thenCombine会将任务的执行结果作为方法入参传递到指定方法中，且该方法有返回值；thenAcceptBoth同样将任务的执行结果作为方法入参，但是无返回值；runAfterBoth没有入参，也没有返回值。注意多个任务中只要有一个执行异常，则将该异常信息作为指定任务的执行结果。allOf是多个任务都执行完成后才会执行，只要有一个任务执行异常，则返回的CompletableFuture执行get方法时会抛出异常，如果都是正常执行，则get返回null。

    @Test
    public void testCompletableFuture() {
        CompletableFuture order = CompletableFuture.supplyAsync(() -> {
            OrderService orderService = new OrderServiceImpl();
            return orderService.queryOrderInfo();
        });
        CompletableFuture trail = CompletableFuture.supplyAsync(() -> {
            TrailService trailService = new TrailServiceImpl();
            return trailService.queryTrail();
        });
        CompletableFuture future = order.thenCombine(trail, (a, b) -> a + b);
        CompletableFuture afterBoth = future.runAfterBoth(trail, () -> {
            System.out.println(future.join());
        });
        CompletableFuture result = CompletableFuture.allOf(afterBoth);
        try {
            System.out.println(result.get());
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
复制代码

applyToEither / acceptEither / runAfterEither/anyOf

这四个方法可以将多个CompletableFuture组合起来，只需要其中一个CompletableFuture执行完成以后，就能执行后面的操作，applyToEither会将已经执行完成的任务的执行结果作为方法入参，并有返回值；acceptEither同样将已经执行完成的任务的执行结果作为方法入参，但是没有返回值；runAfterEither没有方法入参，也没有返回值，注意多个任务中只要有一个执行异常，则将该异常信息作为指定任务的执行结果。anyOf多个任务只要有一个任务执行完成，后续任务就可执行。

    @Test
    public void testCompletableFuture() {
        CompletableFuture order = CompletableFuture.supplyAsync(() -> {
            OrderService orderService = new OrderServiceImpl();
            return orderService.queryOrderInfo();
        });
        CompletableFuture trail = CompletableFuture.supplyAsync(() -> {
            TrailService trailService = new TrailServiceImpl();
            return trailService.queryTrail();
        });
        CompletableFuture future = order.applyToEither(trail, (result) -> result);
        CompletableFuture afterBoth = future.runAfterEither(trail, () -> {
            System.out.println(future.join());
        });
        CompletableFuture result = CompletableFuture.anyOf(afterBoth,order);
        try {
            System.out.println(result.get());
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
复制代码
项目中如何使用
CompletableFuture在自定义线程时，默认使用的线程池是 ForkJoinPool.commonPool()，对于我们用java常做的IO密集型任务，默认线程池是远远不够使用的；在双核及以下机器上，默认线程池又会退化为为每个任务创建一个线程，相当于没有线程池。因此对于CompletableFuture在项目中的使用一定要自定义线程池，同时又要注意自定义线程池，线程池有个容量满了的拒绝策略，如果采用丢弃策略的拒绝策略，并且allOf方法和get方法如果没有设置超时则会无限期的等待下去，接下来我们通过自定义线程使用CompletableFuture。
  1. 自定义线程池，此处通过继承ThreadPoolExecutor，重写了shutdown() 、shutdownNow() 、beforeExecute() 和 afterExecute()方法来统计线程池的执行情况，此处还可以结合Spring和appllo实现自定义扩展线程池，下一篇可以聊聊扩展思路以及实现方案，不同对的业务场景使用的不同的线程池，一是方便出现问题的排查，另外就是类似于Hystrix隔离的方案；
package com.zto.lbd;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 线程池监控类
 *
 * @author wangtongzhou 18635604249
 * @since 2022-02-23 07:27
 */
public class ThreadPoolMonitor extends ThreadPoolExecutor {

    private static final Logger LOGGER = LoggerFactory.getLogger(ThreadPoolMonitor.class);

    /**
     * 保存任务开始执行的时间，当任务结束时，用任务结束时间减去开始时间计算任务执行时间
     */
    private ConcurrentHashMap startTimes;

    /**
     * 线程池名称，一般以业务名称命名，方便区分
     */
    private String poolName;

    /**
     * 调用父类的构造方法，并初始化HashMap和线程池名称
     *
     * @param corePoolSize    线程池核心线程数
     * @param maximumPoolSize 线程池最大线程数
     * @param keepAliveTime   线程的最大空闲时间
     * @param unit            空闲时间的单位
     * @param workQueue       保存被提交任务的队列
     * @param poolName        线程池名称
     */
    public ThreadPoolMonitor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
                             TimeUnit unit, BlockingQueue workQueue, String poolName) {
        this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
                Executors.defaultThreadFactory(), poolName);
    }


    /**
     * 调用父类的构造方法，并初始化HashMap和线程池名称
     *
     * @param corePoolSize    线程池核心线程数
     * @param maximumPoolSize 线程池最大线程数
     * @param keepAliveTime   线程的最大空闲时间
     * @param unit            空闲时间的单位
     * @param workQueue       保存被提交任务的队列
     * @param threadFactory   线程工厂
     * @param poolName        线程池名称
     */
    public ThreadPoolMonitor(int corePoolSize, int maximumPoolSize, long keepAliveTime,
                             TimeUnit unit, BlockingQueue workQueue,
                             ThreadFactory threadFactory, String poolName) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        this.startTimes = new ConcurrentHashMap<>();
        this.poolName = poolName;
    }


    /**
     * 线程池延迟关闭时（等待线程池里的任务都执行完毕），统计线程池情况
     */
    @Override
    public void shutdown() {
        // 统计已执行任务、正在执行任务、未执行任务数量
        LOGGER.info("{} 关闭线程池， 已执行任务: {}, 正在执行任务: {}, 未执行任务数量: {}",
                this.poolName, this.getCompletedTaskCount(), this.getActiveCount(), this.getQueue().size());
        super.shutdown();
    }

    /**
     * 线程池立即关闭时，统计线程池情况
     */
    @Override
    public List shutdownNow() {
        // 统计已执行任务、正在执行任务、未执行任务数量
        LOGGER.info("{} 立即关闭线程池，已执行任务: {}, 正在执行任务: {}, 未执行任务数量: {}",
                this.poolName, this.getCompletedTaskCount(), this.getActiveCount(), this.getQueue().size());
        return super.shutdownNow();
    }

    /**
     * 任务执行之前，记录任务开始时间
     */
    @Override
    protected void beforeExecute(Thread t, Runnable r) {
        startTimes.put(String.valueOf(r.hashCode()), new Date());
    }

    /**
     * 任务执行之后，计算任务结束时间
     */
    @Override
    protected void afterExecute(Runnable r, Throwable t) {
        Date startDate = startTimes.remove(String.valueOf(r.hashCode()));
        Date finishDate = new Date();
        long diff = finishDate.getTime() - startDate.getTime();
        // 统计任务耗时、初始线程数、核心线程数、正在执行的任务数量、
        // 已完成任务数量、任务总数、队列里缓存的任务数量、池中存在的最大线程数、
        // 最大允许的线程数、线程空闲时间、线程池是否关闭、线程池是否终止
        LOGGER.info("{}-pool-monitor: " +
                        "任务耗时: {} ms, 初始线程数: {}, 核心线程数: {}, 正在执行的任务数量: {}, " +
                        "已完成任务数量: {}, 任务总数: {}, 队列里任务数量: {}, 池中存在的最大线程数: {}, " +
                        "最大线程数: {},  线程空闲时间: {}, 线程池是否关闭: {}, 线程池是否终止: {}",
                this.poolName,
                diff, this.getPoolSize(), this.getCorePoolSize(), this.getActiveCount(),
                this.getCompletedTaskCount(), this.getTaskCount(), this.getQueue().size(), this.getLargestPoolSize(),
                this.getMaximumPoolSize(), this.getKeepAliveTime(TimeUnit.MILLISECONDS), this.isShutdown(), this.isTerminated());
    }

    /**
     * 生成线程池所用的线程，改写了线程池默认的线程工厂，传入线程池名称，便于问题追踪
     */
    static class MonitorThreadFactory implements ThreadFactory {
        private static final AtomicInteger poolNumber = new AtomicInteger(1);
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        /**
         * 初始化线程工厂
         *
         * @param poolName 线程池名称
         */
        MonitorThreadFactory(String poolName) {
            SecurityManager s = System.getSecurityManager();
            group = Objects.nonNull(s) ? s.getThreadGroup() : Thread.currentThread().getThreadGroup();
            namePrefix = poolName + "-pool-" + poolNumber.getAndIncrement() + "-thread-";
        }

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(group, r, namePrefix + threadNumber.getAndIncrement(), 0);
            if (t.isDaemon()) {
                t.setDaemon(false);
            }
            if (t.getPriority() != Thread.NORM_PRIORITY) {
                t.setPriority(Thread.NORM_PRIORITY);
            }
            return t;
        }
    }
}
复制代码
  1. 使用自定义线程池的CompletableFuture；
    private final static BlockingQueue workQueue = new ArrayBlockingQueue<>(100);
    private final static ThreadPoolMonitor threadPoolMonitor = new ThreadPoolMonitor(5, 10, 100L,
            TimeUnit.SECONDS, workQueue, "monitor");

    @Test
    public void testCompletableFuture() {
        CompletableFuture order = CompletableFuture.supplyAsync(() -> {
            OrderService orderService = new OrderServiceImpl();
            return orderService.queryOrderInfo();
        },threadPoolMonitor);
        String result=order.join();
        assertTrue(Objects.nonNull(result));
    }
复制代码




        
    
        

            

    






    



            
            
相关推荐

            

                                                

                                                            
[常用工具] git基础学习笔记_git工具有哪些

                    

                        
添加推送信息，-m=messagegitcommit-m“添加注释”查看状态...

                    

                                                        

                                

                                                            
centos7安装部署gitlab_centos7安装git服务器

                    

                        
一、Gitlab介1.1gitlab信息GitLab是利用RubyonRails一个开源的版本管理系统，实现一个自托管的Git项目仓库，可通过Web界面进行访问公开的或者私人项目。...

                    

                                                        

                                

                                                            
太高效了!玩了这么久的Linux，居然不知道这7个终端快捷键

                    

                        
作为Linux用户，大家肯定在Linux终端下敲过无数的命令。有的命令很短，比如：ls、cd、pwd之类，这种命令大家毫无压力。但是，有些命令就比较长了，比如：...

                    

                                                        

                                

                                                            
提高开发速度还能保证质量的10个小窍门

                    

                        
养成坏习惯真是分分钟的事儿，而养成好习惯却很难。我发现，把那些对我有用的习惯写下来，能让我坚持住已经花心思养成的好习惯。...

                    

                                                        

                                

                                                            
版本管理最好用的工具，你懂多少?

                    

                        
版本控制（Revisioncontrol）是一种在开发的过程中用于管理我们对文件、目录或工程等内容的修改历史，方便查看更改历史记录，备份以便恢复以前的版本的软件工程技术。...

                    

                                                        

                                

                                                            
Git回退到某个版本_git回退到某个版本详细步骤

                    

                        
在开发过程，有时会遇到合并代码或者合并主分支代码导致自己分支代码冲突等问题，这时我们需要回退到某个commit_id版本1，查看所有历史版本，获取git的某个历史版本id...

                    

                                                        

                                

                                                            
Kubernetes + Jenkins + Harbor 全景实战手册

                    

                        
Kubernetes+Jenkins+Harbor全景实战手册在现代企业级DevOps体系中，Kubernetes(K8s)、Jenkins和Harbor组成的CI/CD流水...

                    

                                                        

                                

                                                            
git常用命令整理_git常见命令

                    

                        
一、Git仓库完整迁移完整迁移，就是指，不仅将所有代码移植到新的仓库，而且要保留所有的commit记录1.随便找个文件夹，从原地址克隆一份裸版本库...

                    

                                                        

                                

                                                            
第三章:Git分支管理(多人协作基础)

                    

                        
3.1分支基本概念分支是Git最强大的功能之一，它允许你在主线之外创建独立的开发线路，互不干扰。理解分支的工作原理是掌握Git的关键。核心概念：HEAD：指向当前分支的指针...

                    

                                                        

                                

                                                            
云效Codeup怎么创建分支并进行分支管理

                    

                        
云效Codeup怎么创建分支并进行分支管理，分支是为了将修改记录分叉备份保存，不受其他分支的影响，所以在同一个代码库里可以同时进行多个修改。创建仓库时，会自动创建Master分支作为默认分支，后续...

                    

                                                        

                                

                                                            
git 如何删除本地和远程分支?_git怎么删除远程仓库

                    

                        
Git分支对于开发人员来说是一项强大的功能，但要维护干净的存储库，就需要知道如何删除过时的分支。本指南涵盖了您需要了解的有关本地和远程删除Git分支的所有信息。了解Git分支...

                    

                                                        

                                

                                                            
git 实现一份代码push到两个git地址上

                    

                        
一直以来想把自己的博客代码托管到github和coding上想一次更改一次push两个地址一起更新今天有空查资料实践了下本博客的github地址coding的git地址如果是Gi...

                    

                                                        

                                

                                                            
git操作:cherry-pick和rebase_git cherry-pick bad object

                    

                        
在编码中经常涉及到分支之间的代码同步问题，那就需要cherry-pick和rebase命令问题：如何将某个分支的多个commit合并到另一个分支，并在另一个分支只保留一个commit记录解答：假设有两...

                    

                                                        

                                

                                                            
模型文件硬塞进 Git，GitHub 直接打回原形:使用Git-LFS管理大文件

                    

                        
前言最近接手了一个计算机视觉项目代码是屎山就不说了，反正我也不看代码主要就是构建一下docker镜像，测试一下部署的兼容性这本来不难但是，国内服务器的网络环境实在是恶劣，需要配置各种镜像（dock...

                    

                                                        

                                

                                                            
防弹少年团田柾国《Euphoria》2周年 获世界实时趋势榜1位 恭喜呀

                    

                        
当天韩国时间凌晨3时左右，该曲在Twitter上以“2YearsWithEuphoria”的HashTag登上了世界趋势1位。在韩国推特实时趋势中，从上午开始到现在“Euphoria2岁”的Has...

                    

                                                        

                
            


            
        
        

                                    

                
一周热门

                

                    
                

            

                                    




最近发表












标签列表








                    

    







    

                赣ICP备2024040476号-12