Java异步

1、同步、异步、阻塞、非阻塞

理解Java的异步调用，还需要从这四个基础概念理解；如果要从Java方面来理解这四个概念，需要对Java的IO模型有了解，Java 对于 IO 的封装分为 BIO、NIO 和 AIO，BIO 对应的是阻塞同步 IO，NIO 和 AIO 对应的都是非阻塞同步 IO。下面梳理一下对这几个概念的理解。https://www.jianshu.com/p/0efedb229e98

一般的系统无法避免IO依赖，无论是从本地磁盘或从网络读取，是远程通信和本地文件读写的核心。但是由于IO的速度比CPU慢很多，所以IO成了系统的瓶颈，也就很有必要学习IO模型，避免瓶颈，提升性能。

1.1 Linux IO模型

首先从操作系统层面来说，以Linux为例，Linux内核把所有的外来设备当做一个文件，对本地文件有file descriptor处理，对网络文件有socket file descriptor处理，所以可以把二者一起分析。

文件读取设计两个部分，一个是用户进程，一个是内核，用户进程是没有权利操作系统资源的，所以用户进程需要通过系统调用来调用内核，从而达到文件读取的目的。整个过程分三步：

• 用户进程调用read方法向内核发起读请求
• 内核将要读取的信息复制到缓冲区
• 内核将数据从缓冲区复制到用户进程空间

从这个步骤来看，阻塞非阻塞是针对调用方法来说的，如果用户进程在内核将数据准备完成前一致处于阻塞等待状态，就是阻塞的；如果内核还没有准备好，用户调用后直接返回，并没有阻塞，此时就是非阻塞的。。。同步异步是针对内核数据到用户进程这一过程看的，同步是指读写事件就绪后（一般是采用轮询检查），用户进程再自己负责进行读写的操作，内核向用户进程复制数据的过程仍然是阻塞的；如果是内核将数据复制到用户进程，并在复制完成后通知用户，用户没有主动检查数据准备状态，就是异步的，同步和异步的本质区别是，内核数据复制到用户空间时，用户进程是否进行等待。

Linux中用的比较多的是IO多路复用模型，I/O多路复用可以监视多个描述符，一旦某个描述符读写操作就绪，便可以通知程序进行相应的读写操作，这个模型也是同步IO。

1.2 Java IO模型

下面来看Java中的IO模型Java对于IO的封装分为BIO、NIO和AIO。BIO对应的是阻塞同步IO，NIO和AIO对应的都是非阻塞同步IO。

1.2.1 BIO

Java中的BIO对应的是同步阻塞IO。BIO针对每一个客户端请求都建立一个Socket连接，当客户端没有数据时线程一直处于阻塞状态。这种IO模型的优点是简单，但是服务端的线程个数和客户端并发访问数呈正比，如果访问增多，线程也会增多导致OOM；可以通过线程池来解决，但是线程池可以接受的线程数量也是由系统决定的，并且线程上下文切换会导致CPU使用率不高，所以BIO不适合大规模使用。
BIO请求代码如下：

{
 ExecutorService executor = Excutors.newFixedThreadPollExecutor(100);//线程池

 ServerSocket serverSocket = new ServerSocket();
 serverSocket.bind(8088);
 while(!Thread.currentThread.isInturrupted()){//主线程死循环等待新连接到来
 Socket socket = serverSocket.accept();
 executor.submit(new ConnectIOnHandler(socket));//为新的连接创建新的线程
}

class ConnectIOnHandler extends Thread{
    private Socket socket;
    public ConnectIOnHandler(Socket socket){
       this.socket = socket;
    }
    public void run(){
      while(!Thread.currentThread.isInturrupted()&&!socket.isClosed()){死循环处理读写事件
          String someThing = socket.read()....//读取数据
          if(someThing!=null){
             ......//处理数据
             socket.write()....//写数据
          }

      }
    }
}

1.2.2 NIO

NIO对应的是同步非阻塞IO。NIO有几个关键的概念，selector、channel、buffer。

• buffer是包含需要读取或写入的数据的缓冲区。NIO中所有数据的读写均通过缓冲区进行操作。相对于BIO流的好处是可以重复读取，并且可以双向通信。
• channel是一个双向的数据读写的通道，buffer就是建立在channel的基础上流通的。
• selector是NIO的核心，通过不断轮询注册在其之上的channel，当selector发现某个channel有数据状态有变化时，所以只需要一个线程，就可以解决所有的连接请求。

NIO代码如下：

 interface ChannelHandler{
      void channelReadable(Channel channel);
      void channelWritable(Channel channel);
   }
   class Channel{
     Socket socket;
     Event event;//读，写或者连接
   }

   //IO线程主循环:
   class IoThread extends Thread{
   public void run(){
   Channel channel;
   while(channel=Selector.select()){//选择就绪的事件和对应的连接，阻塞的不需要担心cpu空转
      if(channel.event==accept){
         registerNewChannelHandler(channel);//如果是新连接，则注册一个新的读写处理器
      }
      if(channel.event==write){
         getChannelHandler(channel).channelWritable(channel);//如果可以写，则执行写事件
      }
      if(channel.event==read){
          getChannelHandler(channel).channelReadable(channel);//如果可以读，则执行读事件
      }
    }
   }
   Map<Channel，ChannelHandler> handlerMap;//所有channel的对应事件处理器
  }

NIO由原来的阻塞读写（占用线程）变成了单线程轮询事件，找到可以进行读写的网络描述符进行读写。除了事件的轮询是阻塞的（没有可干的事情必须要阻塞），剩余的I/O操作都是纯CPU操作，没有必要开启多线程。

并且由于线程的节约，连接数大的时候因为线程切换带来的问题也随之解决，进而为处理海量连接提供了可能。

1.2.3 AIO

还不会。。。。。

2、Java代码中的异步

上面说的是Java IO模型的同步异步，在业务代码中可不可以体现异步呢？答案是肯定的。

2.1 同步代码

public class Main {                                                                          
    public static void main(String[] args) {                                                 
                                                                                             
        long begin = System.currentTimeMillis();                                             
                                                                                             
        calculate();                                                                         
        long end = System.currentTimeMillis();                                               
        System.out.println(String.format("耗时%ss", String.valueOf(end - begin)));             
    }                                                                                        
                                                                                             
    public static  void calculate(){                                                         
        // 模拟耗时                                                                              
        try {                                                                                
            TimeUnit.SECONDS.sleep(5);                                                       
        } catch (InterruptedException e) {                                                   
            e.printStackTrace();                                                             
        }                                                                                    
                                                                                             
        System.out.println("计算完成");                                                          
    }                                                                                        
}                                                                                            

代码中调用方必须要等到被调用方执行结束后，才可以继续执行。

2.2 基于Future的异步调用

Future 模式相当于一个占位符，代表一个操作的未来的结果，其简单的概念不在本文中介绍，直接给出总结：Future 模式可以细分为将来式和回调式两种模式。

2.2.1 将来式

public class Main {
    public static void main(String[] args) throws ExecutionException, InterruptedException {

        ExecutorService service = Executors.newFixedThreadPool(10);
        long begin = System.currentTimeMillis();
        Future<String> future = service.submit(new Callable<String>() {
            @Override
            public String call() throws Exception {
               return calculate();
            }
        });
        long end = System.currentTimeMillis();
        System.out.println("do otherthing");
        System.out.println(future.get());
        System.out.println(String.format("耗时%ss", String.valueOf(end - begin)));
    }

    public static  String calculate(){
        // 模拟耗时
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return "100";
    }
}

提交任务到线程池后，主线程可以继续做自己的事情，但是future.get()还是阻塞的，所以如果在提交任务后直接get，并不会提高效率，反而由于线程的开销会比同步调用更慢。除了get，还可以通过isDone方法轮询检查是否计算完成，本质上和get没有什么区别。

所以Future虽然可以实现异步，但是还是有缺点，对比JavaScript的异步，不满足一下几点：

• 结果的获取还是通过get阻塞的方式，更好的方式是采用回调
• 如果多个Future形成依赖，会产生回调地狱，如果可以链式调用就更好了。

这种方式在Future的第二种回调式中声明了。

2.2.2回调式

这种方式就是回调，但是Future 并没有实现 callback,addListener 这样的方法，想要在 JAVA 中体验到 callback 的特性，得引入一些额外的框架。有Netty、Guava第三方框架可以使用， jdk1.8 已经提供了一种更为高级的回调方式：CompletableFuture。

2.2.2.1 Guava-ListeningExecutorService

public class Main {
    public static void main(String[] args) throws ExecutionException, InterruptedException {

        long begin = System.currentTimeMillis();
        ListeningExecutorService service = MoreExecutors.listeningDecorator(Executors.newSingleThreadExecutor());
        ListenableFuture<Integer> future = service.submit(new Callable<Integer>() {
            public Integer call() throws Exception {
                System.out.println("执行耗时操作...");
                calculate();
                return 100;
            }
        });

        Futures.addCallback(future, new FutureCallback<Integer>() {

            @Override
            public void onSuccess(Integer result) {
                System.out.println("计算成功:" + result);
            }

            @Override
            public void onFailure(Throwable t) {
                System.out.println("计算失败:" + t.getMessage());
            }
        });

        long end = System.currentTimeMillis();
        System.out.println("do otherthing");
        System.out.println(String.format("耗时%ss", String.valueOf(end - begin)));


    }

    public static  String calculate(){
        // 模拟耗时
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return "100";
    }
}

这下可以彻底不用管结果了，😌😌😌，但是还是有链式调用的问题，所以还是看看CompletableFuture的用法，这是一个非常好用的异步编程类

2.3 JDK-CompletableFuture

2.3.1初始化

有几个常用的初始化静态工厂方法：

CompletableFuture.runAsync(Runnable runnable);
CompletableFuture.runAsync(Runnable runnable, Executor executor);

CompletableFuture.supplyAsync(Supplier<U> supplier);
CompletableFuture.supplyAsync(Supplier<U> supplier, Executor executor)

• runAsync 方法接收的是 Runnable 的实例，意味着它没有返回值
• supplyAsync 方法对应的是有返回值的情况
• 这两个方法的带 executor，表示让任务在指定的线程池中执行，不指定的话，通常任务是在 ForkJoinPool.commonPool() 线程池中执行的，里面是守护线程。

2.3.2 任务执行

//前三个任务 A 无返回值，所以对应的，第 2 行和第 3 行代码中，resultA 其实是 null。
CompletableFuture.runAsync(() -> {}).thenRun(() -> {}); 
CompletableFuture.runAsync(() -> {}).thenAccept(resultA -> {}); 
CompletableFuture.runAsync(() -> {}).thenApply(resultA -> "resultB");

// thenRun(Runnable runnable)，任务 A 执行完执行 B，并且 B 不需要 A 的结果。
CompletableFuture.supplyAsync(() -> "resultA").thenRun(() -> {});
//thenAccept(Consumer action)，任务 A 执行完执行 B，B 需要 A 的结果，但是任务 B 不返回值。
CompletableFuture.supplyAsync(() -> "resultA").thenAccept(resultA -> {});
//thenApply(Function fn)，任务 A 执行完执行 B，B 需要 A 的结果，同时任务 B 有返回值。
CompletableFuture.supplyAsync(() -> "resultA").thenApply(resultA -> resultA + " resultB");

解释在注释里，如果任务 B 后面还有任务 C，往下继续调用 .thenXxx() 即可。

2.3.3 异常处理

public CompletableFuture<T> exceptionally(Function<Throwable, ? extends T> fn);
public <U> CompletionStage<U> handle(BiFunction<? super T, Throwable, ? extends U> fn);

看一个例子：

CompletableFuture.supplyAsync(() -> "resultA")
    .thenApply(resultA -> resultA + " resultB")
    .thenApply(resultB -> resultB + " resultC")
    .thenApply(resultC -> resultC + " resultD");

如果第一个出现了异常，其他的都不能执行，那么我们怎么处理异常呢？看下面的代码，我们在任务 A 中抛出异常，并对其进行处理：

CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
   throw new RuntimeException();
}).exceptionally(ex -> "errorResultA")
  .thenApply(resultA -> resultA + " resultB")
  .thenApply(resultB -> resultB + " resultC")
  .thenApply(resultC -> resultC + " resultD");

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

上面的代码中，任务 A 抛出异常，然后通过 .exceptionally() 方法处理了异常，并返回新的结果，这个新的结果将传递给任务 B。所以最终的输出结果是：

errorResultA resultB resultC resultD

2.3.4 聚合任务

CompletableFuture<String> cfA = CompletableFuture.supplyAsync(() -> "resultA");
CompletableFuture<String> cfB = CompletableFuture.supplyAsync(() -> "resultB");

//thenAcceptBoth 表示后续的处理不需要返回值
cfA.thenAcceptBoth(cfB, (resultA, resultB) -> {});
// thenCombine 表示需要返回值。
cfA.thenCombine(cfB, (resultA, resultB) -> "result A + B");
//不需要result
cfA.runAfterBoth(cfB, () -> {});

2.3.5 取多个任务的结果

// N个Future全部完成才可以
public static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs){...}
// 只需要满足一个就可以
public static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs) {...}

CompletableFuture cfA = CompletableFuture.supplyAsync(() -> "resultA");
CompletableFuture cfB = CompletableFuture.supplyAsync(() -> 123);
CompletableFuture cfC = CompletableFuture.supplyAsync(() -> "resultC");

CompletableFuture<Void> future = CompletableFuture.allOf(cfA, cfB, cfC);
// 所以这里的 join() 将阻塞，直到所有的任务执行结束
future.join();

2.3.6 compose

与thenCombine不同的是，compose的后一个是基于前一个的结果，而thenCombine只是把两个的结果组合在一起。

CompletableFuture参考：https://www.javadoop.com/post/completable-future