当前位置： 首页 > article >正文

【并发编程】异步编程CompletableFuture实战

article 2024/11/13 9:43:05

文章目录

        • 1.CompletableFuture简介
        • 2.CompletableFuture核心API实战
        • 3.CompletableFuture嵌套案例实战
        • 4.合并两个CompletableFuture案例实战
        • 5.多个CompletableFuture任务组合调度实战

1.CompletableFuture简介

在JDK8之前，我们使用的Java多线程变成，主要是 Thread+Runnable 来完成，但是这种方式有个弊端就是没有返回值。如果想要返回值怎么办呢，大多数人就会想到 Callable + Thread 的方式来获取到返回值。

在这里插入图片描述

通过上面的类继承关系图可以知道 CompletableFuture 实现了 Future 接口和 CompletionStage 。因此 CompletableFuture是对 Futrue的功能增强包含了Future的功能。从继承的另一个 CompletionStage 的名称来看完成阶段性的接口。

CompletableFuture的核心用途：

  • 在项目开发中，由于业务规划逻辑的原因，业务需要从多个不同的地方获取数据，
  • 然后汇总处理为最终的结果，再返回给请求的调用方，就是聚合信息处理类的处理逻辑
  • 如果常用串行请求，则接口响应时间长；利用CompletableFuture则可以大大提升性能
  • 针对多任务，需要进行任务编排调度，也可以使用CompletableFuture进行完成

CompletableFuture类实现了Future和CompletionStage接口，相当于一个Task编排工具

  • Future

    • 表示异步计算的结果，它提供了检查计算是否完成的方法，以等待计算的完成
    • 计算完成后只能使用 get 方法来获取结果，有cancel、get、isDone、isCancelled等方法

  • CompletionStage

    • 是Java8新增接口，用于异步执行中的阶段处理，CompletableFuture是其中的一个实现类
    • 对任务处理可以构造一条结果传递链，在结果传递过程中任何一个CompletionStage都可以对结果进行处理
      • 包括异常处理、类型转换，可以构造非常简单的传递链也可以构造很复杂的传递链
    • 几个CompletionStage可以串联起来，一个完成的阶段可以触发下一阶段的执行

  • 当前的Task到底由那个Thread执行，使用的不好可能会有性能问题, 根据CompletableFuture的方法命名可以掌握

    • 不带Async的方法，比如thenAccept，表示该方法将继续在当前执行CompletableFuture的方法线程中执行
    • 带Async的方法，比如thenAcceptAsync，表示异步，在线程池中执行
    • 在没有指定线程池的情况下，使用的是CompletableFuture内部的线程池 ForkJoinPool ,线程数默认是 CPU 的核心数
      • 一般不要所有业务共用一个线程池，避免有任务执行一些很慢的 I/O 操作，
      • 会导致线程池中所有线程都阻塞在 I/O 操作上，从而造成线程饥饿，影响整个系统的性能

方法API

  • CompletableFuture静态方法，执行异步任务的API
//无返回值,默认使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码
public static CompletableFuture<Void>   runAsync(Runnable runnable)

//无返回值,可以自定义线程池
public static CompletableFuture<Void>  runAsync(Runnable runnable, Executor executor)


//有返回值,默认使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码
public static <U> CompletableFuture<U>  supplyAsync(Supplier<U> supplier)

//有返回值,可以自定义线程池
public static <U> CompletableFuture<U>  supplyAsync(Supplier<U> supplier, Executor executor)
  • CompletableFuture对象，获取结果的API
//如果返回值没有返回,一直阻塞
V get()

//设置等待超时的时间
V get(long timeout,Timeout unit);

//有返回值就返回, 线程抛出异常就返回设置的默认值
T getNow(T defaultValue);
  • CompletableFuture对象，其他重点API
//方法无返回值,当前任务正常完成以后执行,当前任务的执行结果可以作为下一任务的输入参数
thenAccept

//方法有返回值,当前任务正常完成以后执行，当前任务的执行的结果会作为下一任务的输入参数
thenApply

//对不关心上一步的计算结果，执行下一个操作
thenRun

2.CompletableFuture核心API实战

（1）supplyAsync方法实战，有返回值,默认使用ForkJoinPool.commonPool() 作为它的线程池执行异步代码。

    public static void testCompletableFuture1() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(()->{
            return "lixiang";
        });
        System.out.println(future.get());
    }

在这里插入图片描述

（2）runAsync方法实战，无返回值

    public static void testCompletableFuture2() throws ExecutionException, InterruptedException {
        CompletableFuture<Void> future = CompletableFuture.runAsync(()->{
            System.out.println("runAsync方法执行。。。");
        });
        System.out.println(future.get());
    }

在这里插入图片描述

（3）thenApply 组合调度，能拿到上步执行的结果，并且当前执行完有任务返回值的

    public static void testCompletableFuture3() throws ExecutionException, InterruptedException {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(()->{
            System.out.println("supplyAsync方法执行。。。");
            return "lixiang";
        }).thenApply((ele)->{
            System.out.println("thenApply方法执行。。。,拿到上一步的执行结果："+ele);
            return ele + " is a java工程师";
        });
        System.out.println(future.get());
    }

在这里插入图片描述

（4）thenAccept 组合调度，能拿到上步执行的结果，当前执行完无任务返回值的

    public static void testCompletableFuture4() throws ExecutionException, InterruptedException {
        CompletableFuture<Void> future = CompletableFuture.supplyAsync(()->{
            System.out.println("supplyAsync方法执行。。。");
            return "lixiang";
        }).thenAccept((ele)->{
            System.out.println("thenAccept方法执行。。。,拿到上一步的执行结果："+ele);
        });
        System.out.println(future.get());
    }

在这里插入图片描述

3.CompletableFuture嵌套案例实战

  • 需求

    • 日常的任务中，通常定义的方法都会返回 CompletableFuture 类型，方便后续操作
    • 然后将该任务的执行结果Future作为方法入参然后执行指定的方法, 返回一个新的CompletableFuture
    • 任务它们之间存在着业务逻辑上的先后顺序

  • thenCompose

    • 用来连接两个CompletableFuture，是生成一个新的CompletableFuture，用于组合多个CompletableFuture
    • 也可以使用 thenApply() 方法来描述关系，但返回的结果就会发生 CompletableFuture 的嵌套
      • CompletableFuture<CompletableFuture< Product >> 这样的情况，需要get两次

（1）编写商品类

@Data
public class Product {

    private int id;

    private String title;

    private String detail;
}

（2）模拟商品service

public class ProductService {

    private static final Map<Integer,String> map = new HashMap<>();
    static {
        map.put(1,"iphone14");
        map.put(2,"iphone 蓝牙耳机");
        map.put(3,"Mac Book Pro-详情图内容");
        map.put(4,"小香风深蓝色大衣");
        map.put(5,"清热解火菊花茶");
        map.put(6,"补肝养肾枸杞大枣茶");
        map.put(7,"颈椎病康复指南");
    }

    public String getById(int id){
        try {
            Thread.sleep(1000);
            System.out.println("ProductService#getById方法运行线程："+Thread.currentThread().getName());
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return map.get(id);
    }

}

（3）模拟商品详情service

public class ProductDetailService {

    private static final Map<Integer,String> map = new HashMap<>();
    static {
        map.put(1,"iphone14-详情图内容");
        map.put(2,"iphone 蓝牙耳机-详情图内容");
        map.put(3,"Mac Book Pro-详情图内容");
        map.put(4,"小香风深蓝色大衣-详情图内容");
        map.put(5,"清热解火菊花茶-详情图内容");
        map.put(6,"补肝养肾枸杞大枣茶-详情图内容");
        map.put(7,"颈椎病康复指南-详情图内容");
    }

    public String getById(int id){
        try {
            Thread.sleep(1000);
            System.out.println("DetailService # getById方法运行线程："+Thread.currentThread().getName());
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return map.get(id);
    }
}

（4）测试方法，正常用thenApply，需要get两次

    public static void test1() throws ExecutionException, InterruptedException {

        ProductService productService = new ProductService();

        ProductDetailService productDetailService = new ProductDetailService();

        //第一步异步 ，第二部异步
        CompletableFuture<CompletableFuture<Product>> future = CompletableFuture.supplyAsync(() -> {
            Product product = new Product();
            String title = productService.getById(1);
            product.setId(product.getId());
            product.setTitle(title);
            return product;
        }).thenApply(product -> CompletableFuture.supplyAsync(() -> {
            String detail = productDetailService.getById(1);
            product.setDetail(detail);
            return product;
        }));

        //这块 获取 商品信息要get两次
        System.out.println("获取商品信息:"+future.get().get());

    }

（5）测试方法，用thenCompose，只需要get一次即可

	public static void test2() throws ExecutionException, InterruptedException {

        ProductService productService = new ProductService();

        ProductDetailService productDetailService = new ProductDetailService();

        //第一步异步 ，第二部异步
        CompletableFuture<Product> compose = CompletableFuture.supplyAsync(() -> {
            Product product = new Product();
            String title = productService.getById(1);
            product.setId(product.getId());
            product.setTitle(title);
            return product;
        }).thenCompose(product -> CompletableFuture.supplyAsync(() -> {
            String detail = productDetailService.getById(1);
            product.setDetail(detail);
            return product;
        }));

        //这块 获取 商品信息要get两次
        System.out.println("获取商品信息:"+compose.get());

    }

在这里插入图片描述

4.合并两个CompletableFuture案例实战

  • 需求

    • 需要请求两个个接口，然后把对应的CompletableFuture进行合并，返回一个新的CompletableFuture

  • thenCombine

    • 在两个任务都执行完成后，把两个任务的结果合并，有返回值。

  • 编码实战

	public static void test3() throws ExecutionException, InterruptedException {
        ProductService productService = new ProductService();
        ProductDetailService detailService = new ProductDetailService();

        int id = 1;
        //第1个任务
        CompletableFuture<Product> baseProductFuture = CompletableFuture.supplyAsync(() -> {
            String title = productService.getById(id);
            Product product = new Product();
            product.setTitle(title);
            product.setId(id);
            return product;
        });

        //第2个任务
        CompletableFuture<Product> detailProductFuture = CompletableFuture.supplyAsync(() -> {
            String detail = detailService.getById(id);
            Product product = new Product();
            product.setDetail(detail);
            product.setId(id);
            return product;
        });

        CompletableFuture<Product> compose = baseProductFuture.thenCombine(detailProductFuture,
                (base, detail) -> {
                    base.setDetail(detail.getDetail());
                    return base;
                });
        System.out.println(compose.get());
    }

在这里插入图片描述

  • thenAccepetBoth
    • 在两个任务都执行完成后，把两个任务的结果合并，有返回值。
  • 编码实战
	public static void test3() throws ExecutionException, InterruptedException {
        ProductService productService = new ProductService();
        ProductDetailService detailService = new ProductDetailService();

        int id = 1;
        //第1个任务
        CompletableFuture<Product> baseProductFuture = CompletableFuture.supplyAsync(() -> {
            String title = productService.getById(id);
            Product product = new Product();
            product.setTitle(title);
            product.setId(id);
            return product;
        });

        //第2个任务
        CompletableFuture<Product> detailProductFuture = CompletableFuture.supplyAsync(() -> {
            String detail = detailService.getById(id);
            Product product = new Product();
            product.setDetail(detail);
            product.setId(id);
            return product;
        });

        CompletableFuture<Void> acceptBoth = baseProductFuture.thenAcceptBoth(detailProductFuture, (base, detail) -> {
            base.setDetail(detail.getDetail());
            System.out.println(base);
        });
        System.out.println(acceptBoth.get());
    }

在这里插入图片描述

5.多个CompletableFuture任务组合调度实战

背景

  • 前面学习处理两个 Future 的关系，如果超过两个Future，如何处理他们的一些聚合关系呢
  • 方法 allOf 和 anyOf
    • 两个函数都是静态函数，参数是变长的 CompletableFuture 的集合，前者是「与」，后者是「或」
    • allOf
      • 返回值是 CompletableFuture< Void >类型
      • 因为allOf没有返回值，所以通过thenApply，获取每个 CompletableFuture 的执行结果
    • anyOf
      • 只要有任意一个 CompletableFuture 结束，就可以做接下来的事情，不像 allOf 要等待所有的 CompletableFuture 结束
      • 每个 CompletableFuture 的返回值类型都可能不同，无法判断是什么类型
      • 所以 anyOf 的返回值是 CompletableFuture< Object >类型

（1）allOf编码实战

	public static void testAllOf() throws Exception {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("future1执行完成");
            return "future1执行完成";
        });


        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("future2执行完成");
            return "future2执行完成";
        });

        CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(5);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("future3执行完成");
            return "future3执行完成";
        });

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

        //阻塞，直到所有任务结束。
        System.out.println(Thread.currentThread().getName() + ":" + LocalDateTime.now() + ":阻塞");
        //调用join方法等待全部任务完成
        all.join();

        if (all.isDone()) {
            //一个需要耗时2秒，一个需要耗时3秒，只有当最长的耗时3秒的完成后，才会结束。
            System.out.println("全部任务执行完成");
        }
        System.out.println(Thread.currentThread().getName() + ":" + LocalDateTime.now() + ":阻塞结束");
    }

在这里插入图片描述

（2）anyOf编码实战

	public static void testAnyOf() throws Exception {
        CompletableFuture<String> future1 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(3);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("future1执行完成");
            return "future1执行完成";
        });


        CompletableFuture<String> future2 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("future2执行完成");
            return "future2执行完成";
        });

        CompletableFuture<String> future3 = CompletableFuture.supplyAsync(() -> {
            try {
                TimeUnit.SECONDS.sleep(5);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("future3执行完成");
            return "future3执行完成";
        });

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

        //阻塞，最快任务执行完成 任务结束。
        System.out.println(Thread.currentThread().getName() + ":" + LocalDateTime.now() + ":阻塞");
        //调用join方法等待最快的一个任务执行
        any.join();

        if (any.isDone()) {
            //一个需要耗时2秒，一个需要耗时3秒，一个耗时5秒 当最短的完成则会结束
            System.out.println(any.get()+"任务执行完成");
        }
        System.out.println(Thread.currentThread().getName() + ":" + LocalDateTime.now() + ":阻塞结束");
    }

在这里插入图片描述


http://www.kler.cn/a/9791.html

相关文章：

  • 河南省的一级科技查新机构有哪些？
  • 三种单例实现
  • 重学SpringBoot3-整合 Elasticsearch 8.x (三)使用Repository
  • 基于Python的膳食健康系统
  • c语言数据结构与算法--简单实现栈和队列的出栈与入栈
  • 什么是数据平台？10 个值得了解的大数据平台示例
  • MySQL开发05-MySQL开发规范
  • 分布式文件系统FastDFS
  • android framework实战车机手机系统开发环境相关问题补充
  • 为什么提升客户服务是长期成功的关键
  • 高精度尘埃粒子计数器工厂空气质量监测必备
  • FreeRTOS 列表和列表项
  • java equals和==的区别
  • 自由度的理解
  • SM59 RFC 目标 SAP_PROXY_ESR 设置到服务资源库连接的检查列表
  • ChatGTP如此强大，我们普通人如何利用它来赚钱？
  • 小程序页面跳转方法
  • SpringSecurity+OAUTH2集成多种登录方式
  • 穿越不确定性，Serverless已翻过千山万水
  • Mybatis（五）------Mybatis执行Mapper接口的方法流程
  • 学习经验分享【NO.23】深度学习目标检测理论知识（持续更新）
  • DETR【论文阅读】
  • 【Android开发经验】-- 如何实现RecyclerView子项的点击事件？
  • 基于springboot+mybatis的图书购物网站
  • 《Java8实战》第4章 引入流
  • 【voliate 关键字】