异步编程

异步编程

 

 

什么是异步化?

 将串行执行的流程任务,拆分成多线程操作,可以实现异步化编程。异步编程,也能使程序更快。

java中的异步编程类CompletableFuture

我们先来看一个例子:

泡茶的过程,我们分了 3 个任务:任务 1 负责洗水壶、烧开水,任务 2 负责洗茶壶、洗茶杯和拿茶叶,任务 3 负责泡茶。其中任务 3 要等待任务 1 和任务 2 都完成后才能开始

异步编程

 

 

 

// 任务 1:洗水壶 -> 烧开水

CompletableFuture<Void> f1 = 
  CompletableFuture.runAsync(()->{
  System.out.println("T1: 洗水壶...");
  sleep(1, TimeUnit.SECONDS);

  System.out.println("T1: 烧开水...");
  sleep(15, TimeUnit.SECONDS);

});

// 任务 2:洗茶壶 -> 洗茶杯 -> 拿茶叶

CompletableFuture<String> f2 = 

  CompletableFuture.supplyAsync(()->{
  System.out.println("T2: 洗茶壶...");
  sleep(1, TimeUnit.SECONDS);

  System.out.println("T2: 洗茶杯...");
  sleep(2, TimeUnit.SECONDS);

  System.out.println("T2: 拿茶叶...");
  sleep(1, TimeUnit.SECONDS);

  return " 龙井 ";

});

// 任务 3:任务 1 和任务 2 完成后执行:泡茶
CompletableFuture<String> f3 = 
  f1.thenCombine(f2, (__, tf)->{
    System.out.println("T1: 拿到茶叶:" + tf);
    System.out.println("T1: 泡茶...");
    return " 上茶:" + tf;

  });

// 等待任务 3 执行结果

System.out.println(f3.join());
void sleep(int t, TimeUnit u) {

  try {

    u.sleep(t);

  }catch(InterruptedException e){}

}

// 一次执行结果:
T1: 洗水壶...
T2: 洗茶壶...
T1: 烧开水...
T2: 洗茶杯...
T2: 拿茶叶...
T1: 拿到茶叶: 龙井
T1: 泡茶...
上茶: 龙井

 

 

创建 CompletableFuture 对象主要靠下面代码中展示的这 4 个静态方法,根据方法传入的参数,有以下区别:

1.Runnable 接口的 run() 方法没有返回值,Supplier 接口的 get() 方法是有返回值的

异步编程

 

 

 

2.上面两个方法的扩展,可以指定线程池。

默认情况下 CompletableFuture 会使用公共的 ForkJoinPool 线程池,这个线程池默认创建的线程数是 CPU 的核数(也可以通过 JVM option:-Djava.util.concurrent.ForkJoinPool.common.parallelism 来设置 ForkJoinPool 线程池的线程数)。如果所有 CompletableFuture 共享一个线程池,那么一旦有任务执行一些很慢的 I/O 操作,就会导致线程池中所有线程都阻塞在 I/O 操作上,从而造成线程饥饿,进而影响整个系统的性能。所以,强烈建议你要根据不同的业务类型创建不同的线程池,以避免互相干扰

异步编程

 

 

 

   对于一个异步操作,你需要关注两个问题:异步操作什么时候结束?  如何获取异步操作的执行结果?

  类图中展示CompletetableFuture类实现了Future、CompletionStage接口,因为实现了 Future 接口, 异步结束可以用,获取结果可以用 isDone(),结果可以用get()来获取。

异步编程

 

 

 

二、万能的CompletionStage 接口

这里还举前面烧水泡茶的例子,其中洗水壶和烧开水就是串行关系,洗水壶、烧开水和洗茶壶、洗茶杯这两组任务之间就是并行关系,而烧开水、拿茶叶和泡茶就是汇聚关系。

异步编程

 

 

 

CompletionStage 接口可以清晰地描述任务之间的这种时序关系,例如前面提到的 f3 = f1.thenCombine(f2, ()->{}) 描述的就是一种汇聚关系。烧水泡茶程序中的汇聚关系是一种 AND 聚合关系,这里的 AND 指的是所有依赖的任务(烧开水和拿茶叶)都完成后才开始执行当前任务(泡茶)。既然有 AND 聚合关系,那就一定还有 OR 聚合关系,所谓 OR 指的是依赖的任务只要有一个完成就可以执行当前任务。

1. 串行关系  

这些都描述的是串行关系

异步编程

以上这些都有什么区别呢?

thenApply 系列函数里参数 fn 的类型是接口 Function<T, R>,这个接口里与 CompletionStage 相关的方法是 R apply(T t),这个方法既能接收参数也支持返回值,所以 thenApply 系列方法返回的是CompletionStage<R>

而 thenAccept 系列方法里参数 consumer 的类型是接口Consumer<T>,这个接口里与 CompletionStage 相关的方法是 void accept(T t),这个方法虽然支持参数,但却不支持回值,所以 thenAccept 系列方法返回的是CompletionStage<Void>

thenRun 系列方法里 action 的参数是 Runnable,所以 action 既不能接收参数也不支持返回值,所以 thenRun 系列方法返回的也是CompletionStage<Void>

这些方法里面 Async 代表的是异步执行 fn、consumer 或者 action。其中,需要你注意的是 thenCompose 系列方法,这个系列的方法会新创建出一个子流程,最终结果和 thenApply 系列是相同的。

 

2. AND 汇聚关系

 异步编程

 

 

3. OR 汇聚关系

异步编程

 

 

下面的示例代码展示了如何使用 applyToEither() 方法来描述一个 OR 汇聚关系

CompletableFuture<String> f1 = 
  CompletableFuture.supplyAsync(()->{
    int t = getRandom(5, 10);
    sleep(t, TimeUnit.SECONDS);
    return String.valueOf(t);
});

 
CompletableFuture<String> f2 = 
  CompletableFuture.supplyAsync(()->{
    int t = getRandom(5, 10);
    sleep(t, TimeUnit.SECONDS);
    return String.valueOf(t);
});


CompletableFuture<String> f3 = 
  f1.applyToEither(f2,s -> s);
System.out.println(f3.join());

 

4. 异常处理

那在异步编程里面,异常该如何处理呢?7/0 这种运行时异常呢?

异步编程

 

 

CompletableFuture<Integer> 
  f0 = CompletableFuture
    .supplyAsync(()->7/0))
    .thenApply(r->r*10)
    .exceptionally(e->0);

System.out.println(f0.join());

 

下面的示例代码展示了如何使用 exceptionally() 方法来处理异常,exceptionally() 的使用非常类似于 try{}catch{}中的 catch{},但是由于支持链式编程方式,所以相对更简单。既然有 try{}catch{},那就一定还有 try{}finally{},whenComplete() 和 handle() 系列方法就类似于 try{}finally{}中的 finally{},无论是否发生异常都会执行 whenComplete() 中的回调函数 consumer 和 handle() 中的回调函数 fn。whenComplete() 和 handle() 的区别在于 whenComplete() 不支持返回结果,而 handle() 是支持返回结果的。

 

切记:

异步编程中一定要考虑到处理异常。

 

引用:部分内容、图片引用王宝令的异步编程。

 

上一篇:多线程(二)


下一篇:CompletableFuture异步编程