前端与后端的异步编排(promise、async、await 、CompletableFuture)

2023-12-22 14:51:11

前端与后端的异步编排

前端和后端都可以进行异步任务,作为前后端都学的我。更得“凑凑热闹”。宝藏文章,不收藏就可惜喽。

1、为什么需要异步编排

异步编程和异步编排在现代编程中变得越来越重要,特别是在处理大规模、高并发、分布式系统等方面。以下是一些主要原因:

  1. 响应性(Responsiveness): 异步编程允许在执行某些耗时操作时保持系统的响应性。在传统的同步编程中,如果一个操作需要花费很长时间,整个应用程序可能会被阻塞,用户体验会受到影响。通过使用异步编程,可以在执行耗时任务的同时继续处理其他事务,使应用程序更加响应。

  2. 性能提升: 异步操作可以有效地提高系统的性能。在某些情况下,通过并行执行异步任务,可以更有效地利用计算资源,加速程序的执行速度。这在处理大规模数据、网络请求等任务时特别有用。

  3. 避免阻塞: 在某些情况下,例如网络请求或文件读写,同步操作可能会导致程序被阻塞,直到操作完成。异步编程可以避免这种阻塞,允许程序在等待结果的同时继续执行其他任务。

  4. 分布式系统: 在分布式系统中,各个组件之间需要进行协同工作。异步编程使得在不同组件之间进行非阻塞通信更为容易,有助于构建高效的分布式系统。

  5. 事件驱动编程: 异步编程很适合事件驱动的场景,其中程序的执行流程由事件的发生和处理来驱动。这种模型通常用于处理用户交互、服务器请求等场景。

  6. 资源利用: 异步编程可以更有效地利用资源,例如在等待IO操作完成时,CPU可以继续执行其他任务,从而提高整体系统的效率。

2、前端中的异步

2.1 、Promise的使用

2.1.1、Promise的基础概念

Promise 在前端通常用于处理异步任务。Promise 是 JavaScript 中的一个对象,用于表示异步操作的最终完成或失败(或者一个异步操作的结果)。

Promise 有三个状态:

  1. Pending(进行中): 初始状态,表示异步操作正在进行中。

  2. Fulfilled(已完成): 表示异步操作已成功完成,并返回一个值。

  3. Rejected(已拒绝): 表示异步操作发生错误,并返回一个原因(错误信息)。

  4. 它只有两种状态可以转化,即

    • 操作成功: pending -> fulfilled
    • 操作失败: pending -> rejected

    注意:并且这个状态转化是单向的,不可逆转,已经确定的状态(fulfilled/rejected)无法转回初始状态(pending)。

Promise 可以通过 .then() 处理已完成的状态,通过 .catch() 处理已拒绝的状态。这种处理方式使得异步操作的代码更加清晰、可读,避免了回调地狱(callback hell)的问题。

例如:

function asyncOperation() {
  return new Promise((resolve, reject) => {
    // 异步操作,比如一个网络请求
    setTimeout(() => {
      const success = true; // 模拟异步操作是否成功
      if (success) {
        resolve("Operation successful"); //修改返回的Promise的状态为Fulfilled
      } else {
        reject("Operation failed");      //修改返回的Promise的状态为Rejected
      }
    }, 1000);
  });
}

asyncOperation()//通过返回的Promise对象进行调用
  .then(result => {
    console.log(result); // 处理异步操作成功的情况(状态为Fulfilled)
  })
  .catch(error => {
    console.error(error); // 处理异步操作失败的情况(状态为Rejected)
  });

在上述例子中,asyncOperation 函数返回一个 Promise 对象,通过 .then() 处理成功情况,通过 .catch() 处理失败情况。这种方式更容易理解和维护,尤其是在处理多个异步操作时。

2.1.2、Promise中的两个回调函数
  1. Promise.prototype.then(callback)

    Promise对象含有then方法,then()调用后返回一个Promise对象,意味着实例化后的Promise对象可以进行链式调用,而且这个**then()方法可以接收两个函数,**一个是处理成功后的函数,一个是处理错误结果的函数。

var promise1 = new Promise(function(resolve, reject) {
  // 2秒后置为接收状态
  setTimeout(function() {
    resolve('success');     //通过修改Promise的状态来触发函数回调
  }, 2000);
});

promise1.then(function(data) {      //注意这里的then()的括号,里面就是成功、失败的回调函数
  console.log(data); // success
}, function(err) {              
  console.log(err); // 没有异常,不执行
}).then(function(data) {
  // 上一步的then()方法没有返回值
  console.log('链式调用:' + data); // 链式调用:undefined 
}).then(function(data) {
  // ....
});

其实就是根据上一个Promise中的resolve(‘success’); 或者reject(“Operation failed”); 修改Promise后续的函数回调,注意所谓的函数回调一般是异步的,就是你给回调函数一个触发状态之后你就可以去干自己的事了,后续会异步根据触发状态来触发对应的回调函数。

2.Promise.prototype.catch(callback)

catch()方法和then()方法一样(这两个更像是并列的关系,注意例子中的括号),都会返回一个新的Promise对象,它主要用于捕获异步操作时出现的异常。因此,我们通常省略then()方法的第二个参数,把错误处理控制权转交给其后面的catch()函数,.catch() 可以添加在 Promise 链的任何地方,而不仅仅是在链的最后。一个 Promise 链中可以有多个 .catch() 来处理不同位置的错误。

function asyncOperation() {
  return new Promise((resolve, reject) => {
    // 模拟异步操作
    setTimeout(() => {
      const success = Math.random() < 0.5; // 模拟成功或失败
      if (success) {
        resolve("Operation successful");
      } else {
        reject("Operation failed");
      }
    }, 1000);
  });
}

asyncOperation()
  .then(result => {
    console.log(result); // 处理成功的情况(注意这里就省略了.then里面的第二个回调函数)
    // 这里抛出一个错误
    throw new Error("Custom error");
  })
  .catch(error => {
    console.error("Caught an error:", error); // 处理错误的情况
  })
  .then(() => {
    console.log("After catch"); // 即使前面有错误,仍然会执行这里
  });
2.1.3、工具方法
1、Promise.all()

Promise.all() 是一个用于处理多个 Promise 并发执行的工具方法。它接收一个包含多个 Promise 的可迭代对象(比如数组),并返回一个新的 Promise,该 Promise 在所有输入的 Promise 都成功(resolved)时才会成功,如果任何一个 Promise 失败(rejected),它就会失败,返回失败的那个 Promise 的结果。

使用 Promise.all() 的典型场景是在需要同时发起多个异步请求,等待所有请求都完成后再执行一些操作。

const promise1 = new Promise((resolve, reject) => {
  setTimeout(() => resolve('Promise 1'), 1000);
});

const promise2 = new Promise((resolve, reject) => {
  setTimeout(() => resolve('Promise 2'), 2000);
});

const promise3 = new Promise((resolve, reject) => {
  setTimeout(() => resolve('Promise 3'), 1500);
});

Promise.all([promise1, promise2, promise3])  //注意这里一般是传入一个promise数组
  .then(results => {
    console.log('All promises resolved:', results);
    // 所有 Promise 都成功时的操作
  })
  .catch(error => {
    console.error('At least one promise rejected:', error);
    // 如果有任何一个 Promise 失败,这里处理错误
  });

2、Promise.race()

Promise.race() 是另一个用于处理多个 Promise 的工具方法,但它与 Promise.all() 不同。Promise.race() 接收一个包含多个 Promise 的可迭代对象(比如数组),并返回一个新的 Promise,该 Promise 在输入的 Promise 中有一个率先完成(无论是成功还是失败)时就会完成。

使用 Promise.race() 的典型场景是在需要多个异步操作中只关注最先完成的那个。

const promise1 = new Promise((resolve, reject) => {
  setTimeout(() => resolve('Promise 1'), 1000);
});

const promise2 = new Promise((resolve, reject) => {
  setTimeout(() => resolve('Promise 2'), 2000);
});

const promise3 = new Promise((resolve, reject) => {
  setTimeout(() => resolve('Promise 3'), 1500);
});

Promise.race([promise1, promise2, promise3])
  .then(winner => {
    console.log('The first promise resolved:', winner);
    // 最先完成的 Promise 的操作
  })
  .catch(error => {
   console.error('The first promise that failed:', error);
    // 值得注意的是,上面的数组中所有promise只要有一个promise率先失败,整个 Promise.race() 就会失败,进入 .catch() 部分,打印出 率先失败的结果。
  });
3、Promise.resolve()

Promise.resolve() 是一个用于创建一个已完成(fulfilled)状态的 Promise 的静态方法。它返回一个 Promise 对象,可以包装一个已经存在的值或者另一个 Promise 对象。

Promise.resolve() 有几种使用方式:

  1. 返回一个已解决的 Promise:

    const resolvedPromise = Promise.resolve("Resolved value");
    

    这个例子中,resolvedPromise 是一个已完成状态的 Promise,其值为字符串 “Resolved value”。

  2. 包装一个普通值:

    const valuePromise = Promise.resolve(42);
    

    这里,valuePromise 是一个已完成状态的 Promise,其值为数字 42。

  3. 包装另一个 Promise:

    const anotherPromise = new Promise((resolve, reject) => {
      // 模拟异步操作,这里是一个立即拒绝的 Promise
      reject("Another Promise rejected");
    });
    
    const wrappedPromise = Promise.resolve(anotherPromise); //返回的promise与包装的anotherPromise状态一致,都是reject
    
    

wrappedPromise
.then(value => {
console.log(“Resolved:”, value);
})
.catch(error => {
console.error(“Rejected:”, error); //会执行这里的拒绝的方法
});


当使用 `Promise.resolve()` 包装另一个 Promise 时,返回的 Promise 的状态(fulfilled 或 rejected)将取决于被包装的 Promise 的状态。如果被包装的 Promise 处于已解决状态(fulfilled),那么返回的 Promise 也将处于已解决状态;如果被包装的 Promise 处于拒绝状态(rejected),那么返回的 Promise 也将处于拒绝状态。

##### 4**、Promise.reject()**

`Promise.reject()` 与 `Promise.resolve()` 不同,它不能用于包装另一个 Promise。`Promise.reject()` 直接返回一个处于拒绝状态的 Promise,而不考虑其他 Promise 对象的状态。

?```javascript
const rejectedPromise = Promise.reject("Rejected for a reason");

rejectedPromise
.then(value => {
 console.log("Resolved:", value);
})
.catch(error => {
 console.error("Rejected:", error);  //执行catch方法
});

2.2 、async 与 await 的使用

es7新增的 async函数可以更舒适地与promise协同工作,它叫做async/await,它是非常的容易理解和使用。

asyncawait 是 JavaScript 中用于处理异步操作的关键字,它们通常用于简化 Promise 的使用。下面是它们的基本用法:

  1. async 函数:

    • async 它被放置在一个函数前面,用于定义一个异步函数。在异步函数内部,你可以使用 await 来等待其他异步操作promise的完成。
    • 异步函数总是返回一个 Promise 对象。(这里得注意)
    async function myAsyncFunction() {
      // 异步操作
      return result;
    }
    
  2. await 表达式:

    • await 用于等待一个 Promise 对象的解决或拒绝,并返回 Promise 的结果。
    • 在使用 await 的地方,代码将等待异步操作完成后再继续执行。
    async function f() {
        let promise = new Promise((resolve, reject) => {
            setTimeout(() => resolve('done!'), 1000)
        })
        let result = await promise // 直到promise返回一个resolve值(*)
     alert(result) // 'done!' 
    }
    

f()


在上述例子中,会在1s后输出'done!'

3. **处理错误:**

- 使用 `try...catch` 来捕获异步操作中的错误。`catch` 部分将捕获 `try` 部分中抛出的异常。

```javascript
async function example() {
  try {
    const result = await someAsyncFunction();
    console.log(result);
  } catch (error) {
    console.error("An error occurred:", error);
  }
}

如果 someAsyncFunction 返回一个拒绝状态的 Promise,那么控制流将跳到 catch 部分,捕获错误。

  1. 并发执行:

    • 使用 Promise.all() 或其他并发执行的方法来同时执行多个异步操作。
    async function example() {
      const promise1 = someAsyncFunction1();
      const promise2 = someAsyncFunction2();
      const promise3 = someAsyncFunction3();
    
      try {
        const results = await Promise.all([promise1, promise2, promise3]);
        console.log('All promises resolved:', results);
        // 所有 Promise 都成功时的操作
      } catch (error) {
        console.error('At least one promise rejected:', error);
        // 如果有任何一个 Promise 失败,这里处理错误
      }
    }
    
    // 调用示例函数
    example();
    
  2. 异步函数总是返回一个 Promise 对象:

    • 例如,下面的代码返回resolved值为1的promise,我们可以测试一下:
async function f() {
    return 1
}
f().then(alert) // 弹出1

我们也可以显式的返回一个promise,这个将会是同样的结果

async function f() {
    return Promise.resolve(1)
}
f().then(alert) // 弹出1

3、后端中的异步

3.1 CompletableFuture的使用

CompletableFuture是jdk8的新特性。CompletableFuture实现了CompletionStage接口和Future接口,前者是对后者的一个扩展,增加了异步会点、流式处理、多个Future组合处理的能力,使Java在处理多任务的协同工作时更加顺畅便利。

3.2、创建异步任务

3.2.1. supplyAsync

supplyAsync是创建带有返回值的异步任务。它有如下两个方法,一个是使用默认线程池(ForkJoinPool.commonPool())的方法,一个是带有自定义线程池的重载方法

// 带返回值异步请求,默认线程池
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)
 
// 带返回值的异步请求,可以自定义线程池
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

这段代码是Java中的CompletableFuture类的supplyAsync方法的签名,该方法用于异步执行一个Supplier,并返回一个CompletableFuture对象,代表异步计算的结果。

public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 自定义线程池
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
            System.out.println("do something....");
            return "result";
        }, executorService);
 
        //等待子任务执行完成
        System.out.println("结果->" + cf.get());
}
扩展:CompletableFuture中的方法签名:

public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

  • public static <U>:这表示这是一个泛型方法,其中的<U>是类型参数,是方法声明的一部分,表示这是一个泛型方法。这个 <U> 是一个类型参数,它是一个占位符,代表一种未知的类型。当你调用这个方法时,你可以用具体的类型替换这个 U,以满足实际需求。
  • CompletableFuture<U>:这是方法的返回类型,代表一个CompletableFuture对象,该对象最终会包含异步计算的结果。后面的这个U,它是返回的 CompletableFuture 包含的值的类型。
  • supplyAsync:这是方法的名称,表示它用于执行一个供应商(Supplier)的异步计算。
  • (Supplier<U> supplier, Executor executor):这是方法的参数列表。
    • Supplier<U>:是一个函数式接口,它代表一个不接受任何参数但返回类型为U的函数。在这里,它表示提供异步计算结果的函数。
    • Executor executor:是一个用于执行计算的Executor。Executor负责管理线程池,决定异步计算是在哪个线程上执行。
什么是泛型方法?

泛型方法是一种在方法中使用泛型类型参数的方法。在Java中,你可以为一个方法定义泛型类型,这使得该方法能够在调用时接受不同类型的参数,提高了代码的灵活性和重用性。

泛型方法的语法格式如下:

public <T> returnType methodName(T parameter) {
    // 方法体
}

其中:

  • <T> 表示这是一个泛型方法,T 是类型参数的名称,你可以使用任何合法的标识符代表类型参数。
  • returnType 是方法的返回类型。
  • methodName 是方法的名称。
  • (T parameter) 是方法的参数列表,其中 T 是类型参数。

下面是一个简单的示例,演示了如何编写和调用泛型方法:

public class GenericMethodExample {
    // 泛型方法,接受一个参数并返回
    public <T> T printAndReturn(T value) {
        System.out.println("Input value: " + value);
        return value;
    }

    public static void main(String[] args) {
        GenericMethodExample example = new GenericMethodExample();

        // 调用泛型方法,传入不同类型的参数
        String stringValue = example.printAndReturn("Hello, Generics!");
        Integer intValue = example.printAndReturn(42);

        System.out.println("Returned String: " + stringValue);
        System.out.println("Returned Integer: " + intValue);
    }
}

在这个例子中,printAndReturn 方法是一个泛型方法,可以接受不同类型的参数。通过使用泛型方法,我们可以在编写代码时更好地支持不同类型的数据,而不必为每个类型编写相似的方法。

什么是供给型接口和消费型接口?

其实对于学习Completable很重要的一点就是看它方法参数是供给型接口和消费性接口

在Java中,供给型接口和消费型接口是Java函数式编程中的两个常见类型。它们都是函数式接口的一种,函数式接口是只有一个抽象方法的接口。Java中的函数式接口可以用Lambda表达式或方法引用来创建实例。

  1. 供给型接口(Supplier):

    • Supplier 是一个提供(供给)值的函数式接口。
    • 它定义了一个名为 get 的抽象方法,该方法不接受任何参数,返回一个值。
    • Supplier 接口通常用于表示那些无需输入参数,但需要产生一个结果的场景。
    • 示例:
      Supplier<String> supplier = () -> "Hello, World!";
      String result = supplier.get(); // 获取供给的值
      
  2. 消费型接口(Consumer):

    • Consumer 是一个消费值的函数式接口。
    • 它定义了一个名为 accept 的抽象方法,该方法接受一个参数,但没有返回值(返回类型为 void)。
    • Consumer 接口通常用于表示那些需要对输入进行处理但不产生结果的场景。
    • 示例:
      Consumer<String> consumer = message -> System.out.println(message);
      consumer.accept("Hello, World!"); // 消费输入值
      
3.2.2、runAsync

runAsync是创建没有返回值的异步任务。它有如下两个方法,一个是使用默认线程池(ForkJoinPool.commonPool())的方法,一个是带有自定义线程池的重载方法

// 不带返回值的异步请求,默认线程池
public static CompletableFuture<Void> runAsync(Runnable runnable)
 
// 不带返回值的异步请求,可以自定义线程池
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)

测试代码:

public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 自定义线程池
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        CompletableFuture<Void> cf = CompletableFuture.runAsync(() -> {
            System.out.println("do something....");   //注意这里并没有返回值
        }, executorService);
 
        //等待任务执行完成
        System.out.println("结果->" + cf.get());
}

3.3、获取任务结果的方法

// 如果完成则返回结果,否则就抛出具体的异常
public T get() throws InterruptedException, ExecutionException 

// 最大时间等待返回结果,否则就抛出具体异常
public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException

// get()方法会抛出checked exception,即必须在方法签名中声明或捕获异常。join()方法会抛出uncheck exception,即RuntimeException,不需要在方法签名中声明或捕获异常。
public T join()

// 如果完成则返回结果值(或抛出任何遇到的异常),否则返回给定的 valueIfAbsent。
public T getNow(T valueIfAbsent)

// 如果任务没有完成,返回的值设置为给定值
public boolean complete(T value)

// 如果任务没有完成,就抛出给定异常
public boolean completeExceptionally(Throwable ex) 

代码示例:

import java.util.concurrent.CompletableFuture;

public class CompletableFutureExample {
    public static void main(String[] args) {
        CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "Hello, CompletableFuture!";
        });

        // 使用get()方法获取结果,必须对异常进行处理
        try {
            String result = future.get();
            System.out.println("Result from get(): " + result);
        } catch (Exception e) {
            e.printStackTrace();
        }

        // 使用join()方法获取结果
        String result = future.join();
        System.out.println("Result from join(): " + result);
    }
}

3.4、异步回调处理

3.4.1、thenApply和thenApplyAsync

thenApply 表示某个任务执行完成后执行的动作,即回调方法,会将该任务的执行结果即方法返回值作为入参传递到回调方法中,带有返回值。

public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " cf1 do something....");
            return 1;
        });
        //thenApplyAsync
        CompletableFuture<Integer> cf2 = cf1.thenApplyAsync((result) -> {
            System.out.println(Thread.currentThread() + " cf2 do something....");
            result += 2;
            return result;
        });
        //等待任务1执行完成
        System.out.println("cf1结果->" + cf1.get());
        //等待任务2执行完成
        System.out.println("cf2结果->" + cf2.get());
}
 
public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " cf1 do something....");
            return 1;
        });
        //thenApply
        CompletableFuture<Integer> cf2 = cf1.thenApply((result) -> {
            System.out.println(Thread.currentThread() + " cf2 do something....");
            result += 2;
            return result;
        });
        //等待任务1执行完成
        System.out.println("cf1结果->" + cf1.get());
        //等待任务2执行完成
        System.out.println("cf2结果->" + cf2.get());
}
  • thenApply和thenApplyAsync区别在于,使用thenApply方法时子任务与父任务使用的是同一个线程,而thenApplyAsync在子任务中是另起一个线程执行任务,并且thenApplyAsync可以自定义线程池,默认的使用ForkJoinPool.commonPool()线程池。
  • 虽然 apply 方法有一个参数,但在 thenApply 的用法中,它仍然符合供给型接口的概念,因为它的输入是上一个阶段的结果,而不是外部传递的值。在这种上下文中,Function 接口可以被视为一种供给型接口,因为它提供了一个计算结果的操作。
3.4.2、thenAccept和thenAcceptAsync

thenAccep表示某个任务执行完成后执行的动作,即回调方法,会将该任务的执行结果即方法返回值作为入参传递到回调方法中,无返回值。

public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " cf1 do something....");
            return 1;
        });
       //thenAccept 有参数无返回值  
        CompletableFuture<Void> cf2 = cf1.thenAccept((result) -> {
            System.out.println(Thread.currentThread() + " cf2 do something....");
        });
 
        //等待任务1执行完成
        System.out.println("cf1结果->" + cf1.get());
        //等待任务2执行完成
        System.out.println("cf2结果->" + cf2.get());
}
 
 
public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " cf1 do something....");
            return 1;
        });
        //有参数无返回值
        CompletableFuture<Void> cf2 = cf1.thenAcceptAsync((result) -> {
            System.out.println(Thread.currentThread() + " cf2 do something....");
        });
 
        //等待任务1执行完成
        System.out.println("cf1结果->" + cf1.get());
        //等待任务2执行完成
        System.out.println("cf2结果->" + cf2.get());
}
3.4.3、thenRun和thenRunAsync

thenRun表示某个任务执行完成后执行的动作,即回调方法,无入参,无返回值。 区别还是和之前的一样,是否可能使用新的线程执行异步任务。

public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " cf1 do something....");
            return 1;
        });
        //thenRun
        CompletableFuture<Void> cf2 = cf1.thenRun(() -> {
            System.out.println(Thread.currentThread() + " cf2 do something....");
        });
 
        //等待任务1执行完成
        System.out.println("cf1结果->" + cf1.get());
        //等待任务2执行完成
        System.out.println("cf2结果->" + cf2.get());
}
 
public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " cf1 do something....");
            return 1;
        });
        //thenRunAsync
        CompletableFuture<Void> cf2 = cf1.thenRunAsync(() -> {
            System.out.println(Thread.currentThread() + " cf2 do something....");
        });
 
        //等待任务1执行完成
        System.out.println("cf1结果->" + cf1.get());
        //等待任务2执行完成
        System.out.println("cf2结果->" + cf2.get());
}
3.4.4、whenComplete和whenCompleteAsync

whenCompletewhenCompleteAsync 都是 CompletableFuture 类中的方法,用于注册一个回调以处理异步操作的结果和异常。它们之间的主要区别在于回调的执行方式。

  1. whenComplete 方法:

    • whenComplete 方法注册一个回调函数,该函数会在异步操作完成时执行,不关心之前的计算是成功还是失败。
    • 回调函数的签名为 (result, throwable),其中 result 是计算成功的结果(如果有的话),throwable 是抛出的异常(如果有的话)。
    • 这个回调函数会在执行线程上执行,而不是使用额外的线程池。
    • 示例:
      CompletableFuture<Integer> cf = CompletableFuture.supplyAsync(() -> 42);
      
      cf.whenComplete((result, throwable) -> {
          if (result != null) {
              System.out.println("Result: " + result);
          } else {
              System.err.println("Exception: " + throwable);
          }
      });
      
  2. whenCompleteAsync 方法:

    • whenCompleteAsync 方法与 whenComplete 类似,但它允许你指定一个 Executor,用于执行回调函数。
    • 这个方法的优势在于它可以在指定的线程池中执行回调,而不是在执行线程上执行。
    • 示例:
      CompletableFuture<Integer> cf = CompletableFuture.supplyAsync(() -> 42);
      
      cf.whenCompleteAsync((result, throwable) -> {
          if (result != null) {
              System.out.println("Result: " + result);
          } else {
              System.err.println("Exception: " + throwable);
          }
      }, Executors.newFixedThreadPool(3));
      

总体而言,whenCompletewhenCompleteAsync 提供了一种在异步计算完成时处理结果和异常的机制,让你能够以更灵活的方式管理异步操作。选择使用哪一个取决于你的需求,以及是否需要在特定的线程池中执行回调。

3.4.5、handle和handleAsync

handlehandleAsyncCompletableFuture 类中的方法,用于注册一个回调以处理异步操作的结果和异常,类似于 whenCompletewhenCompleteAsync。它们也有类似于 thenApplythenApplyAsync 的对应方法。但是它们有返回值。

whenCompletewhenCompleteAsync 也允许你指定一个 Executor,但与 handle 不同的是,它们使用默认的 ForkJoinPool.commonPool() 执行回调函数。

  1. handle 方法:

    • handle 方法注册一个回调函数,该函数会在异步操作完成时执行,不关心之前的计算是成功还是失败。
    • 回调函数的签名为 (result, throwable),其中 result 是计算成功的结果(如果有的话),throwable 是抛出的异常(如果有的话)。
    • 这个回调函数会在执行线程上执行,而不是使用额外的线程池。
    • 示例:
      CompletableFuture<Integer> cf = CompletableFuture.supplyAsync(() -> 42);
      
      cf.handle((result, throwable) -> {
          if (result != null) {
              return result + 2;
          } else {
              return 0; // 处理异常,返回默认值
          }
      });
      
  2. handleAsync 方法:

    • handleAsync 方法与 handle 类似,但它允许你指定一个 Executor,用于执行回调函数。
    • 这个方法的优势在于它可以在指定的线程池中执行回调,而不是在执行线程上执行。
    • 示例:
      CompletableFuture<Integer> cf = CompletableFuture.supplyAsync(() -> 42);
      
      cf.handleAsync((result, throwable) -> {
          if (result != null) {
              return result + 2;
          } else {
              return 0; // 处理异常,返回默认值
          }
      }, Executors.newFixedThreadPool(3));
      

3.5、多任务组合处理

3.5.1、thenCombine、thenAcceptBoth 和runAfterBoth

这三个方法都是将两个CompletableFuture组合起来处理,只有两个任务都正常完成时,才进行下阶段任务。

区别:thenCombine会将两个任务的执行结果作为所提供函数的参数,且该方法有返回值;thenAcceptBoth同样将两个任务的执行结果作为方法入参,但是无返回值;runAfterBoth没有入参,也没有返回值。注意两个任务中只要有一个执行异常,则将该异常信息作为指定任务的执行结果。

public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " cf1 do something....");
            return 1;
        });
 
        CompletableFuture<Integer> cf2 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " cf2 do something....");
            return 2;
        });
        //thenCombine 有参数有返回值
        CompletableFuture<Integer> cf3 = cf1.thenCombine(cf2, (a, b) -> {
            System.out.println(Thread.currentThread() + " cf3 do something....");
            return a + b;
        });
 
        System.out.println("cf3结果->" + cf3.get());  //结果是3
}
 
 public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " cf1 do something....");
            return 1;
        });
 
        CompletableFuture<Integer> cf2 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " cf2 do something....");
            return 2;
        });
        //thenAcceptBoth 有参数但是没有返回值
        CompletableFuture<Void> cf3 = cf1.thenAcceptBoth(cf2, (a, b) -> {
            System.out.println(Thread.currentThread() + " cf3 do something....");
            System.out.println(a + b);
        });
 
        System.out.println("cf3结果->" + cf3.get()); //无返回值,所以结果是null
}

public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<Integer> cf1 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " cf1 do something....");
            return 1;
        });
 
        CompletableFuture<Integer> cf2 = CompletableFuture.supplyAsync(() -> {
            System.out.println(Thread.currentThread() + " cf2 do something....");
            return 2;
        });
        //runAfterBoth 无参数无返回值
        CompletableFuture<Void> cf3 = cf1.runAfterBoth(cf2, () -> {
            System.out.println(Thread.currentThread() + " cf3 do something....");
        });
 
        System.out.println("cf3结果->" + cf3.get()); //这里自然也是null
}
3.5.2、applyToEither、acceptEither和runAfterEither

这三个方法和上面一样也是将两个CompletableFuture组合起来处理,当有一个任务正常完成时,就会进行下阶段任务。

区别:applyToEither会将已经完成任务的执行结果作为所提供函数的参数,且该方法有返回值;acceptEither同样将已经完成任务的执行结果作为方法入参,但是无返回值;runAfterEither没有入参,也没有返回值。

主要就是先来后到,

public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println(Thread.currentThread() + " cf1 do something....");
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "cf1 任务完成";
        });
 
        CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println(Thread.currentThread() + " cf2 do something....");
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "cf2 任务完成";
        });
        //applyToEither 有参数有返回值
        CompletableFuture<String> cf3 = cf1.applyToEither(cf2, (result) -> {
            System.out.println("接收到" + result);
            System.out.println(Thread.currentThread() + " cf3 do something....");
            return "cf3 任务完成";
        });
 
        System.out.println("cf3结果->" + cf3.get());
}
 
运行结果: (因为是有参有返回值,而且是先来后到)
cf2 do something....
cf1 do something....
接收到cf1 任务完成
cf3 do something....
cf3结果 cf3 任务完成

public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println(Thread.currentThread() + " cf1 do something....");
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "cf1 任务完成";
        });
 
        CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println(Thread.currentThread() + " cf2 do something....");
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return "cf2 任务完成";
        }); 
        //acceptEither有参无返回值
        CompletableFuture<Void> cf3 = cf1.acceptEither(cf2, (result) -> {
            System.out.println("接收到" + result);
            System.out.println(Thread.currentThread() + " cf3 do something....");
        });
 
        System.out.println("cf3结果->" + cf3.get());
}
 运行结果:  (有参无返回值,先来后到,所以最后cf3没有返回值是null)
 cf2 do something....
 cf1 do something....
 接收到cf1 任务完成
 cf3 do something....
 cf3结果->null
 
public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println(Thread.currentThread() + " cf1 do something....");
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("cf1 任务完成");
            return "cf1 任务完成";
        });
 
        CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println(Thread.currentThread() + " cf2 do something....");
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("cf2 任务完成");
            return "cf2 任务完成";
        });
        //runAfterEither 无参数无返回值
        CompletableFuture<Void> cf3 = cf1.runAfterEither(cf2, () -> {
            System.out.println(Thread.currentThread() + " cf3 do something....");
            System.out.println("cf3 任务完成");
        });
 
        System.out.println("cf3结果->" + cf3.get());
}
 运行结果:  (无参无返回值,先来后到,所以接受的cf1的参数以及最后cf3返回值都是null)
 cf2 do something....
 cf1 do something....
 接收到cf1 null
 cf3 do something....
 cf3结果->null
3.5.3、allOf 、anyOf

allOf:allOf是多个任务都执行完成后才会执行,只有有一个任务执行异常,则返回的CompletableFuture执行get方法时会抛出异常,如果都是正常执行,则get返回null ,也就是CompletableFuture Void。

anyOf :CompletableFuture是多个任务只要有一个任务执行完成,则返回的CompletableFuture执行get方法时会抛出异常,如果都是正常执行,则get返回首先执行完成任务的结果。

public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println(Thread.currentThread() + " cf1 do something....");
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("cf1 任务完成");
            return "cf1 任务完成";
        });
 
        CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println(Thread.currentThread() + " cf2 do something....");
                int a = 1/0;
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("cf2 任务完成");
            return "cf2 任务完成";
        });
 
        CompletableFuture<String> cf3 = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println(Thread.currentThread() + " cf3 do something....");
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("cf3 任务完成");
            return "cf3 任务完成";
        });
        //allOf 如果都是正常执行,则get返回null,重点在于get是个阻塞的方法,会等待所有的任务完成。
        CompletableFuture<Void> cfAll = CompletableFuture.allOf(cf1, cf2, cf3);
        System.out.println("cfAll结果->" + cfAll.get());
}
执行结果:
cf2 do something....
cf3 do something....
cf1 do something.... 
cf1 任务完成  // 任务正常执行,所以System.out.println("cf1 任务完成");这句话会打印
cf3 任务完成
抛出算数异常,没有结果

public static void main(String[] args) throws ExecutionException, InterruptedException {
        CompletableFuture<String> cf1 = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println(Thread.currentThread() + " cf1 do something....");
                Thread.sleep(2000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("cf1 任务完成");
            return "cf1 任务完成";
        });
 
        CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println(Thread.currentThread() + " cf2 do something....");
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("cf2 任务完成");
            return "cf2 任务完成";
        });
 
        CompletableFuture<String> cf3 = CompletableFuture.supplyAsync(() -> {
            try {
                System.out.println(Thread.currentThread() + " cf3 do something....");
                Thread.sleep(3000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println("cf3 任务完成");
            return "cf3 任务完成";
        });
        //anyOf 如果都是正常执行,则get返回首先执行完成任务的结果。
        CompletableFuture<Object> cfAll = CompletableFuture.anyOf(cf1, cf2, cf3);
        System.out.println("cfAll结果->" + cfAll.get());
        
执行结果:
cf2 do something....
cf3 do something....
cf1 do something.... 
cf1 任务完成    //值得注意的是 “cf3 任务完成” 这句话与“cf1 任务完成”两个不会同时打印,因为anyOf只会阻塞到获取一个任务的结果,然后继续执行,程序退出
cfAll结果->cf1 任务完成
}

好啦,如果能细心看完这篇文章肯定能受益匪浅吧,整理了一上午了,希望对你有帮助吧。

文章来源:https://blog.csdn.net/qq_45925197/article/details/135150473
本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。