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

大家好,我是 方圆。最近在开发 延保服务 频道页时,为了提高查询效率,使用到了多线程技术。为了对多线程方案设计有更加充分的了解,在业余时间读完了《图解 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 模式 结合,以查询订单的任务为例,我们可以将该任务封装为“命令”对象的形式,执行时为每个线程提交一个命令,实现解耦并提高扩展性。在命令模式中,命令对象需要 支持撤销和重做,那么这便在查询出现异常时,提供了补偿处理的可能,命令模式类图关系如下:

在这里插入图片描述

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 并发编程中有着重要的应用。以下是这三个方法的常见使用场景:

  1. complete(T value): 此方法用于显式地完成一个 CompletableFuture,并设置它的结果值。这在你需要在某个计算完成时,手动设置 CompletableFuture 的结果值的场景中非常有用。例如,你可能在一个异步操作完成时,需要设置 CompletableFuture 的结果值。
CompletableFuture<String> future = new CompletableFuture<>();
// Some asynchronous operation
future.complete("Operation Result");
  1. completeExceptionally(Throwable ex): 此方法用于显式地以异常完成一个 CompletableFuture。这在你需要在某个计算失败时,手动设置 CompletableFuture 的异常的场景中非常有用。例如,你可能在一个异步操作失败时,需要设置 CompletableFuture 的异常。
CompletableFuture<String> future = new CompletableFuture<>();
// Some asynchronous operation
future.completeExceptionally(new RuntimeException("Operation Failed"));
  1. 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);

欢迎大家在京东APP内搜索 “京东延保” 跳转延保服务页~

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.mfbz.cn/a/712773.html

如若内容造成侵权/违法违规/事实不符,请联系我们进行投诉反馈qq邮箱809451989@qq.com,一经查实,立即删除!

相关文章

几种经典查找算法

几种经典查找算法 顺序查找法二分查找法判定树 二叉查找树&#xff08;BST&#xff09;索引查找B-树B树散列表&#xff08;hash&#xff09;查找 顺序查找法 顺序查找的平均查找长度为&#xff1a; 时间复杂度为0&#xff08;n&#xff09;&#xff1b; 二分查找法 int bin…

CNN学习(7):用C++实现简单不同参数的卷积模型

目录 一、参数说明和计算公式 1、符号约定 2、输出大小计算公式 二、不同类型的卷积 1、输入3*3*1&#xff0c;卷积核3*3*1&#xff0c;输出1*1*1 &#xff08;1&#xff09;实现代码 &#xff08;2&#xff09;代码说明 2、输入4*4*1&#xff0c;卷积核3*3*1&#xff…

环保评A的意义与价值

环保评A&#xff0c;这个看似简单的称谓&#xff0c;背后却蕴藏着深厚的环保理念和实践标准。在当今社会&#xff0c;环保已经成为一项全球性的议题&#xff0c;各国都在努力推动绿色发展&#xff0c;实现可持续发展目标。那么&#xff0c;环保评A究竟是全国性的认证还是地方性…

Java SSTI服务端模版注入漏洞原理与利用

文章目录 前言Velocity基础语法基础示例命令执行 靶场实践漏洞代码漏洞验证检测工具 FreeMarker基础示例漏洞示例CMS案例 Thymeleaf基础示例漏洞示例安全方案 总结 前言 SSTI&#xff08;Server Side Template Injection&#xff09;全称服务端模板注入漏洞&#xff0c;在 Jav…

开放式耳机值得入手买吗?可以对比这几款开放式耳机看看

居家办公时&#xff0c;选择一款合适的耳机能够有效地提高工作效率。入耳式耳机虽然能够有效地隔绝外界噪音&#xff0c;但长时间佩戴会对耳朵造成负担&#xff0c;甚至引发耳道感染。而头戴式耳机虽然能够提供更好的音质&#xff0c;但体积较大&#xff0c;佩戴起来不够灵活。…

PyTorch -- Batch Normalization(BN) 快速实践

Batch Normalization 可以 改善梯度消失/爆炸问题&#xff1a;前面层的梯度经过多次传递后会变得非常小(大)&#xff0c;从而导致网络收敛速度慢(不收敛)&#xff0c;应用 BN 可缓解加速网络收敛&#xff1a;BN 使得每个神经元的输入分布更加稳定减少过拟合&#xff1a;BN 可减…

求导,积分

求导公式&#xff1a; 复合函数求导法则&#xff1a;两个函数导函数的乘积. 例如&#xff1a;f(x)2x1,f(x)2,g(x)x^24x4,g(x)2x4 那么复合函数&#xff1a; g(f(x))(2x1)^24(2x1)4 把&#xff08;2x1&#xff09;看做整体,则g2(2x1)4 然后再求&#xff08;2x1&#xff09;的导函…

LeetCode | 2879.显示前三行

在 pandas 中&#xff0c;可以使用 head() 方法来读取 DataFrame 的前几行数据。如果想读取指定数量的行&#xff0c;可以在 head() 方法中传入一个参数 n&#xff0c;读取前 n 行 import pandas as pddef selectFirstRows(employees: pd.DataFrame) -> pd.DataFrame:retur…

Dictionary 字典

文章目录 一、什么是字典1.1 字典的创建方式 一、什么是字典 字典&#xff1a; 用来存储数据&#xff0c;与列表和元组不一样的是&#xff0c;字典以键值对的形式对数据进行存储&#xff0c;也就是 key 和 value。相当于 Java 中的 Map。 注意&#xff1a; 1、 key 的值不可重…

C++进阶(一)

个人主页&#xff1a;PingdiGuo_guo 收录专栏&#xff1a;C干货专栏 前言 本篇博客是讲解函数的重载以及引用的知识点的。 文章目录 前言 1.函数重载 1.1何为函数重载 1.2函数重载的作用 1.3函数重载的实现 2.引用 2.1何为引用 2.2定义引用 2.3引用特性 2.4常引用 2…

认识一些分布函数-Frechet分布及其应用

1. 何为Frechet分布 Frechet分布也称为极值分布(EVD)类型II,用于对数据集中的最大值进行建模。它是四种常用极值分布之一。另外三种是古贝尔分布、威布尔分布和广义极值分布(Gumbel Distribution, the Weibull Distribution and the Generalized Extreme Value Distributi…

34 Debian如何配置ELK群集

作者:网络傅老师 特别提示:未经作者允许,不得转载任何内容。违者必究! Debian如何配置ELK群集 《傅老师Debian知识库系列之34》——原创 ==前言== 傅老师Debian知识库特点: 1、拆解Debian实用技能; 2、所有操作在VMware虚拟机实测完成; 3、致力于最终形成Debian知识手…

LVS-DR模式详解:提升网站性能的最佳解决方案

LVS-DR模式原理 用户请求到达Director Server&#xff1a; 用户请求到达Director Server&#xff08;负载均衡服务器&#xff09;&#xff0c;数据包首先到达内核空间的PREROUTING链。数据包源IP&#xff1a;CIP&#xff0c;目标IP&#xff1a;VIP&#xff0c;源MAC&#xff1a…

【内存管理之C语言数组】

1.栈空间上的C数组 糟糕的可用性&#xff0c;但是你将在遗留代码中见到它们 相同类型的对象的内存块 大小必须是常量表达式 第一个元素索引为0 2.指针和C数组 更奇怪的是&#xff1a;数组标识符退化为指向第一个元素的指针 3.访问数组 4.堆空间上的C数组 相同类型的对象的内…

数据库开发——并发控制(第十一章)

文章目录 前言并发执行例题一、封锁二、封锁协议三、可串行调度四、总结 学习目标&#xff1a;重点为并发控制的基本概念及几个基本协议 前言 数据库管理系统必须提供并发控制机制&#xff0c;保证事务的隔离性和一致性 并发执行例题 一、封锁 排他锁称为写锁&#xff0c;共…

智能化状态管理:自动状态流转处理模块

目录 基本背景介绍 具体实现 基本数据准备 基本数据表 状态转换常量 状态转换注解 任务处理模版 各任务实现逻辑 开启比对任务进行处理 降噪字段处理任务处理 开启业务数据比对处理 业务数据比对处理 开始核对数据生成最终报告处理 核对数据生成最终报告处理 状…

小红书教程简化版,从0开始走向专业,小红书-主理人培养计划 (13节)

课程目录 1-小红书分析与拆解.mp4 2-小红书电商玩法.mp4 3-小红书基础信息设置10_1.mp4 4-小红书如何开店&#xff1f;.mp4 5-小红书店铺设置&#xff08;1&#xff09;.mp4 5-小红书店铺设置.mp4 6-小红书笔记制作与产品发布.mp4 7-小红书运营的文案与标题.mp4 8-小红…

Spring Boot 自定义Starter

自定义starter 创建pom项目 <?xml version"1.0" encoding"UTF-8"?> <project xmlns"http://maven.apache.org/POM/4.0.0"xmlns:xsi"http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation"http://maven.ap…

MySQL的三种重要的日志

日志 Mysql有三大日志系统 Undo Log&#xff08;回滚日志&#xff09;&#xff1a;记录修改前的数据&#xff0c;用于事务回滚和 MVCC&#xff08;多版本并发控制&#xff09;。 Redo Log&#xff08;重做日志&#xff09;&#xff1a;记录数据变更&#xff0c;用于崩溃恢复&…

XAMPP PHP-CGI 远程代码执行漏洞(CVE-2024-4577)

漏洞概述&#xff1a; PHP 是一种被广泛应用的开放源代码的多用途脚本语言&#xff0c;PHP-CGI 是 PHP 自带的 FastCGI 管理器。是一个实现了 CGI 协议的程序&#xff0c;用来解释 PHP 脚本的程序&#xff0c;2024 年 6 月 7 日&#xff0c;推特安全上 orange 公开了其漏洞细节…