从多线程设计模式到对 CompletableFuture 的应用

京东云开发者
• 阅读 195

最近在开发 延保服务 频道页时,为了提高查询效率,使用到了多线程技术。为了对多线程方案设计有更加充分的了解,在业余时间读完了《图解 Java 多线程设计模式》这本书,觉得收获良多。本篇文章将介绍其中提到的 Future 模式,以及在实际业务开发中对该模式的应用,而这些内容对于本书来说只是冰山一角,还是推荐大家有时间去阅读原书。

1. Future 模式:“先给您提货单”

我们先来看一个场景:假如我们去蛋糕店买蛋糕,下单后,店员会递给我们提货单并告知“请您傍晚来取蛋糕”。到了傍晚我们拿着提货单去取蛋糕,店员会先和我们说“您的蛋糕已经做好了”,然后将蛋糕拿给我们。

如果将下单蛋糕到取蛋糕的过程抽象成一个方法的话,那么意味着这个方法需要花很长的时间才能获取执行结果,与其一直等待结果,不如先拿着一张“提货单” ,到我们需要取货的时候,再通过它去取,而获取“提货单”的过程是几乎不耗时的,而这个提货单对象就被称为 Future,后续便可以通过它来获取方法的返回值。用 Java 来表示这个过程的话,需要使用到 FutureTaskCallable 两个类,如下:

public class Example {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        // 预定蛋糕,并定义“提货单”
        System.out.println("我:预定蛋糕");
        FutureTask<String> future = new FutureTask<>(() -> {
            System.out.println("店员:请您傍晚来取蛋糕");
            Thread.sleep(2000);
            System.out.println("店员:您的蛋糕已经做好了");

            return "Holiland";
        });
        // 开始做蛋糕
        new Thread(future).start();

        // 去做其他事情
        Thread.sleep(1000);
        System.out.println("我:忙碌中...");
        // 取蛋糕
        System.out.println("我:取蛋糕 " + future.get());
    }
}

// 运行结果:
// 我:预定蛋糕
// 店员:请您傍晚来取蛋糕
// 我:忙碌中...
// 店员:您的蛋糕已经做好了
// 我:取蛋糕 Holiland

方法的调用者可以将任务交给其他线程去处理,无需阻塞等待方法的执行,这样调用者便可以继续执行其他任务,并能通过 Future 对象获取执行结果。

它的运行原理如下:创建 FutureTask 实例时,Callable 对象会被传递给构造函数,当线程调用 FutureTaskrun 方法时,Callable 对象的 call 方法也会被执行。调用 call 方法的线程会同步地获取结果,并通过 FutureTaskset 方法来记录结果对象,如果 call 方法执行期间发生了异常,则会调用 setException 方法记录异常。最后,通过调用 get 方法获取方法的结果,注意这里可能会抛出方法执行时产生的异常

    public void run() {
        // ...
        try {
            // “提货任务”
            Callable<V> c = callable;
            if (c != null && state == NEW) {
                V result;
                boolean ran;
                try {
                    // 调用 callable 的 call 方法
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    // 捕获并设置异常
                    setException(ex);
                }
                if (ran)
                    // 为结果赋值
                    set(result);
            }
        } finally {
            // ...
        }
    }

    protected void set(V v) {
        if (STATE.compareAndSet(this, NEW, COMPLETING)) {
            // 将结果赋值给 outcome 全局变量,供 get 时获取
            outcome = v;
            // 修改状态为 NORMAL
            STATE.setRelease(this, NORMAL); // final state
            finishCompletion();
        }
    }

    protected void setException(Throwable t) {
        if (STATE.compareAndSet(this, NEW, COMPLETING)) {
            // 将异常赋值给 outcome 变量,供 get 时抛出
            outcome = t;
            // 修改状态为 EXCEPTIONAL
            STATE.setRelease(this, EXCEPTIONAL); // final state
            finishCompletion();
        }
    }

    public V get() throws InterruptedException, ExecutionException {
        int s = state;
        // 未完成时阻塞等一等
        if (s <= COMPLETING)
            s = awaitDone(false, 0L);
        return report(s);
    }

    private V report(int s) throws ExecutionException {
        Object x = outcome;
        // 正常结束的话能正常获取到结果
        if (s == NORMAL)
            return (V)x;
        // 否则会抛出异常,注意如果执行中出现异常,调用 get 时会被抛出
        if (s >= CANCELLED)
            throw new CancellationException();
        throw new ExecutionException((Throwable)x);
    }

现在对 Future 模式 已经有了基本的了解:它通过 Future 接口来表示未来的结果,实现 调用者与执行者之间的解耦提高系统的吞吐量和响应速度,那在实践中对该模式是如何使用的呢?

2. 对 Future 模式的实践

因为 延保服务 频道页访问量大且对接口性能要求较高,单线程处理并不能满足性能要求,所以应用了 Future 模式 来提高查询效率,但是并没有借助上文所述的 FutureTask 来实现,而是使用了 CompletableFuture 工具类,它们的实现原理基本一致,但是后者提供的方法和对 链式编程 的支持使代码更加简洁,实现更加容易(相关 API 参考见文末)。

如下是使用 CompletableFuture 异步多线程查询订单列表的逻辑,根据配置的 pageNo 分多条线程查询各页的订单数据:

        List<OrderListInfo> result = new ArrayList<>();
        // 并发查询订单列表
        List<CompletableFuture<List<OrderListInfo>>> futureList = new ArrayList<>();
        try {
            // 配置需要查询的页数 pageNo,并发查询不同页码的订单
            for (int i = 1; i <= pageNo; i++) {
                int curPageNo = i;
                CompletableFuture<List<OrderListInfo>> future = CompletableFuture.supplyAsync(
                        () -> getOrderInfoList(userNo, curPageNo), threadPoolExecutor);

                futureList.add(future);
            }
            // 等待所有线程处理完毕,并封装结果值
            for (CompletableFuture<List<OrderListInfo>> future : futureList) {
                result.addAll(future.get());
            }
        } catch (Exception e) {
            log.error("并发查询用户订单信息异常", e);
        }

这段代码中对异常的处理能进行优化:第 15 行代码,如果某条线程查询订单列表时发生异常,那么在调用 get 方法时会抛出该异常,被 catch 后返回空结果,即使有其他线程查询成功,这些订单结果值也会被忽略掉,可以针对这一点进行优化,如下:

        List<OrderListInfo> result = new ArrayList<>();
        // 并发查询订单列表
        List<CompletableFuture<List<OrderListInfo>>> futureList = new ArrayList<>();
        try {
            // 配置需要查询的页数 pageNo,并发查询不同页码的订单
            for (int i = 1; i <= pageNo; i++) {
                int curPageNo = i;
                CompletableFuture<List<OrderListInfo>> future = CompletableFuture
                        .supplyAsync(() -> getOrderInfoList(userNo, curPageNo), threadPoolExecutor)
                // 添加异常处理
                .exceptionally(e -> {
                    log.error("查询用户订单信息异常", e);
                    return Collections.emptyList();
                });

                futureList.add(future);
            }
            // 等待所有线程处理完毕,并封装结果值
            for (CompletableFuture<List<OrderListInfo>> future : futureList) {
                result.addAll(future.get());
            }
        } catch (Exception e) {
            log.error("并发查询用户订单信息异常", e);
        }

优化后针对查询发生异常的任务打印异常日志,并返回空集合,这样即使单线程查询失败,也不会影响到其他线程查询成功的结果。

CompletableFuture 还提供了 allOf 方法,它返回的 CompletableFuture 对象在所有 CompletableFuture 执行完成时完成,相比于对每个任务都调用 get 阻塞等待任务完成的实现可读性更好,改造后代码如下:

        List<OrderListInfo> result = new ArrayList<>();
        // 并发查询订单列表
        CompletableFuture<List<OrderListInfo>>[] futures = new CompletableFuture[pageNo];
        // 配置需要查询的页数 pageNo,并发查询不同页码的订单
        for (int i = 1; i <= pageNo; i++) {
            int curPageNo = i;
            CompletableFuture<List<OrderListInfo>> future = CompletableFuture
                    .supplyAsync(() -> getOrderInfoList(userNo, curPageNo), threadPoolExecutor)
                    // 添加异常处理
                    .exceptionally(e -> {
                        log.error("查询用户订单信息异常", e);
                        return Collections.emptyList();
                    });

            futures[i - 1] = future;
        }

        try {
            // 等待所有线程处理完毕
            CompletableFuture.allOf(futures).get();
            for (CompletableFuture<List<OrderListInfo>> future : futures) {
                List<OrderListInfo> orderInfoList = future.get();
                if (CollectionUtils.isEmpty(orderInfoList)) {
                    result.addAll(orderInfoList);
                }
            }
        } catch (Exception e) {
            log.error("处理用户订单结果信息异常", e);
        }

Tips: CompletableFuture 的设计初衷是支持异步编程,所以应尽量避免在CompletableFuture 链中使用 get()/join() 方法,因为这些方法会阻塞当前线程直到CompletableFuture 完成,应该在必须使用该结果值时才调用它们。

相关的模式:命令模式

命令模式能将操作的调用者和执行者解耦,它能很容易的与 Future 模式 结合,以查询订单的任务为例,我们可以将该任务封装为“命令”对象的形式,执行时为每个线程提交一个命令,实现解耦并提高扩展性。在命令模式中,命令对象需要 支持撤销和重做,那么这便在查询出现异常时,提供了补偿处理的可能,命令模式类图关系如下:

从多线程设计模式到对 CompletableFuture 的应用

3.《图解Java多线程设计模式》书籍推荐

我觉得本书算得上是一本老书:05 年出版的基于 JDK1.5 的Java多线程书籍,相比于目前我们常用的 JDK1.8 和时髦的 JDK21,在读之前总会让人觉得有一种过时的感觉。但是当我读完时,发现其中的模式能对应上代码中的处理逻辑:对 CompletableFuture 的使用正对应了其中的 Future 模式(异步获取其他线程的执行结果)等等,所以我觉得模式的应用不会局限于技术的新老,它是在某种情况下,研发人员共识或通用的解决方案,在知晓某种模式,采用已有的技术实现它是容易的,而反过来在只掌握技术去探索模式是困难且没有方向的。

同时,我也在考虑一个问题:对于新人学习多线程技术来说,究竟适不适合直接从模式入门呢?因为我对设计模式有了比较多的实践经验,所以对“模式”相关的内容足够敏感,如果新人没有这些经验的话,这对他们来说会不会更像是一个个知识点的堆砌呢?好在的是,本书除了模式相关的内容,对基础知识也做足了铺垫,而且提出的关于多线程编程的思考点也是非常值得参考和学习的,以线程互斥和协同为例,书中谈到:在对线程进行互斥处理时需要考虑 “要保护的东西是什么” ,这样便能够 清晰的确定锁的粒度;对于线程的协同,书中提到的是需要考虑 “放在中间的东西是什么” ,直接的抛出这个观点是不容易理解的,“中间的东西”是在多线程的 生产者和消费者模式 中提出的,部分线程负责生产,生产完成后将对象放在“中间”,部分线程负责消费,消费时取的便是“中间”的对象,而合理规划这些中间的东西便能 消除生产者和消费者之间的速度差异,提高系统的吞吐量和响应速度。而再深入考虑这两个角度时,线程的互斥和协同其实是内外统一的:为了让线程协调运行,必须执行互斥处理,以防止共享的内容被破坏,而线程的互斥是为了线程的协调运行才进行的必要操作。


附:CompletableFuture 常用 API

使用 supplyAsync 方法异步执行任务,并返回 CompletableFuture 对象

如下代码所示,调用 CompletableFuture.supplyAsync 静态方法异步执行查询逻辑,并返回一个新的 CompletableFuture 对象

CompletableFuture<List<Object>> future = CompletableFuture.supplyAsync(() -> doQuery(), executor);

使用 join 方法阻塞获取完成结果

如下代码所示,在封装结果前,调用 join 方法阻塞等待获取结果

futureList.forEach(CompletableFuture::join);

它与 get 方法的主要区别在于,join 方法抛出的是未经检查的异常 CompletionException,并将原始异常作为其原因,这意味着我们可以不需要在方法签名中声明它或在调用 join 方法的地方进行异常处理,而 get 方法会抛出 InterruptedExceptionExecutionException 异常,我们必须对它进行处理,get 方法源码如下:

    public T get() throws InterruptedException, ExecutionException {
        Object r;
        if ((r = result) == null)
            r = waitingGet(true);
        return (T) reportGet(r);
    }

用 thenApply(Function) 和 thenAccept(Consumer) 等回调函数处理结果

如下是使用 thenApply() 方法对 CompletableFuture 的结果进行转换的操作:

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello")
    .thenApply(greeting -> greeting + " World");

使用 exceptionally() 处理 CompletableFuture 中的异常

CompletableFuture 提供了exceptionally() 方法来处理异常,这是一个非常重要的步骤。如果在 CompletableFuture 的运行过程中抛出异常,那么这个异常会被传递到最终的结果中。如果没有适当的异常处理,那么在调用 get()join() 方法时可能会抛出异常。

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    if (true) {
        throw new RuntimeException("Exception occurred");
    }
    return "Hello, World!";
}).exceptionally(e -> "An error occurred");

使用 allOf() 和 anyOf() 处理多个 CompletableFuture

如果有多个 CompletableFuture 需要处理,可以使用 CompletableFuture.allOf() 或者 CompletableFuture.anyOf()allOf() 在所有的 CompletableFuture 完成时完成,而 anyOf() 则会在任意一个 CompletableFuture 完成时完成。

complete()、completeExceptionally()、cancel() 方法

CompletableFuture 的运行是在调用了 complete()completeExceptionally()cancel() 等方法后才会被标记为完成。如果没有正确地完成 CompletableFuture,那么在调用 get() 方法时可能会永久阻塞。这三个方法在 Java 并发编程中有着重要的应用。以下是这三个方法的常见使用场景:

  • complete(T value): 此方法用于显式地完成一个 CompletableFuture,并设置它的结果值。这在你需要在某个计算完成时,手动设置 CompletableFuture 的结果值的场景中非常有用。例如,你可能在一个异步操作完成时,需要设置 CompletableFuture 的结果值。
CompletableFuture<String> future = new CompletableFuture<>();
// Some asynchronous operation
future.complete("Operation Result");
  • completeExceptionally(Throwable ex): 此方法用于显式地以异常完成一个 CompletableFuture。这在你需要在某个计算失败时,手动设置 CompletableFuture 的异常的场景中非常有用。例如,你可能在一个异步操作失败时,需要设置 CompletableFuture 的异常。
CompletableFuture<String> future = new CompletableFuture<>();
// Some asynchronous operation
future.completeExceptionally(new RuntimeException("Operation Failed"));
  • cancel(boolean mayInterruptIfRunning): 此方法用于取消与 CompletableFuture 关联的计算。这在你需要取消一个长时间运行的或者不再需要的计算的场景中非常有用。例如,你可能在用户取消操作或者超时的情况下,需要取消 CompletableFuture 的计算。
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
    // Long running operation
});
// Some condition
future.cancel(true);

这些方法都是线程安全的,可以从任何线程中调用。

使用 thenCompose() 处理嵌套的 CompletableFuture

如果在处理 CompletableFuture 的结果时又创建了新的CompletableFuture,那么就会产生嵌套的 CompletableFuture。这时可以使用 thenCompose() 方法来避免 CompletableFuture 的嵌套,如下代码所示:

CompletableFuture<String> completableFuture
  = CompletableFuture.supplyAsync(() -> "Hello")
    .thenCompose(s -> CompletableFuture.supplyAsync(() -> s + " World"));

使用 thenCombine() 处理两个 CompletableFuture 的结果

CompletableFuture<String> completableFuture
  = CompletableFuture.supplyAsync(() -> "Hello")
    .thenCombine(CompletableFuture.supplyAsync(() -> " World"), (s1, s2) -> s1 + s2);
点赞
收藏
评论区
推荐文章
徐小夕 徐小夕
3年前
《前端实战总结》之迭代器模式的N+1种应用场景
眼看12月就来了,抓住今年的尾巴,好好总结一下前端的不足与收获。这篇文章是笔者写设计模式专题的第二篇文章,也是基于工作中的总结和提炼,在实际应用场景中都会大量使用,至于为什么要写设计模式,主要是为了提高团队代码质量和可维护性,后续会继续推出设计模式相关的文章,供大家参考和学习。你将学到迭代器模式的含义实现一个数组迭代器实现一个对象迭代器
责任链和策略设计模式-基于Java编程语言
责任链和策略设计模式这两种设计模式非常实用,下面简单介绍一下我对这两种设计模式的理解和它们在Spring框架源码中的应用。
Wesley13 Wesley13
3年前
java 泛型详解
对java的泛型特性的了解仅限于表面的浅浅一层,直到在学习设计模式时发现有不了解的用法,才想起详细的记录一下。本文参考java泛型详解、Java中的泛型方法、java泛型详解1\.概述泛型在java中有很重要的地位,在面向对象编程及各种设计模式中有非常广泛的应用。什么是泛型?为什么要使用泛型?泛型,即“参数化类型”。一提到参数,最熟
Wesley13 Wesley13
3年前
java中饿汉与懒汉的故事(单例设计模式)
java中的单例设计模式关于设计模式,这其实是单独存在的东西,它不属于java,但是在java中使用较多,所以今天我就给大家介绍下单例设计模式中的饿汉和懒汉这俩朴素的打工人。首先我先说明下单例设计模式是啥(如果不想了解,可以直接划下去看饿汉和懒汉):类的单例设计模式就是采用一定的方法保证在整个软件系统中,对某个类只能存在一
Stella981 Stella981
3年前
IOS之UIWebView的使用
刚接触IOS开发1年多,现在对于混合式移动端开发越来越流行,因为开发成本上、速度上都比传统的APP开发要好,混合式开发是传统模式与PC网页端相结合的模式。那么提到了APP的混合模式开发,在Android开发中有WebView作为混合模式开发的桥梁,当然在IOS中也同样有一个UIWebView组件来作为混合模式开发的桥梁,那么下面就对UIWebVi
Stella981 Stella981
3年前
Kubernetes operator 模式开发实践
0\.前言近日我们在开发符合我们业务自身需求的微服务平台时,使用了Kubernetes的OperatorPattern来实现其中的运维系统,在本文,我们将实现过程中积累的主要知识点和技术细节做了一个整理。读者在阅读完本文之后,会对OperatorPattern有一个基本的了解,并能将该模式应用到自己的业务中去。除此
Stella981 Stella981
3年前
Reactor中的Thread和Scheduler
简介今天我们要介绍的是Reactor中的多线程模型和定时器模型,Reactor之前我们已经介绍过了,它实际上是观察者模式的延伸。所以从本质上来说,Reactor是和多线程无关的。你可以把它用在多线程或者不用在多线程。今天将会给大家介绍一下如何在Reactor中使用多线程和定时器模型。Thread多线程先看一下之前
Stella981 Stella981
3年前
React 世界的一等公民
Choerodon猪齿鱼平台使用React作为前端应用框架,对前端的展示做了一定的封装和处理,并配套提供了前端组件库ChoerodonUI。结合实际业务情况,不断对组件优化设计,提高代码质量。本文将结合Choerodon猪齿鱼平台使用案例,简单说明组件的分类、设计原则和设计模式,帮助开发者在不同场景下选择正确的设计和方案编写组件(示例代码基于ES6
京东云开发者 京东云开发者
6个月前
工程中实践的微服务设计模式
最近在读《微服务架构设计模式》,开始的时候我非常的好奇,因为在我印象中,设计模式是常说的那23种设计模式,而微服务的设计模式又是什么呢?这个问题也留给大家,在文末我会附上我对这个问题的理解。本次文章的内容主要是工作中对微服务设计模式的应用,希望能对大家有所
京东云开发者 京东云开发者
1个月前
还在自己实现责任链?我建议你造轮子之前先看看这个开源项目
1.前言设计模式在软件开发中被广泛使用。通过使用设计模式,开发人员可以更加高效地开发出高质量的软件系统,提高代码的可读性、可维护性和可扩展性。责任链模式是一种常用的行为型设计模式,它将请求沿着处理链进行发送,直到其中一个处理者对请求进行处理为止。在责任链模