记一次生产中使用CompletableFuture遇到的坑

为什么运用CompletableFuture

事务功用描绘：有一个功用是需求调用基础渠道接口组装我们需求的数据，在这个功用里面我们要调用屡次基础渠道的接口，我们的入参是一个id，但是这个id是一个调集。我们都是运用RPC调用，一般常规的想法去遍历循环这个idList，但是呢这个id调集里面的数据或许会有500个左右。说多不多，说少也不少，首要是在for循环里面屡次去RPC调用是一件特别费时的事情。

我用代码大致描绘一下这个需求：

public List<BasicInfo> buildBasicInfo(List<Long> ids) {
    List<BasicInfo> basicInfoList = new ArrayList<>();
    for (Long id : ids) {
      getBasicData(basicInfoList, id);
     }
   }
​
  private List<BasicInfo> getBasicData(List<BasicInfo> basicInfoList, Long id) {
    BasicInfo basicInfo = rpcGetBasicInfo(id);
    return basicInfoList.add(basicInfo);
   }
​
  public BasicInfo rpcGetBasicInfo(Long id) {
    // 第一次RPC 调用
     rpcInvoking_1()...........
​
    // 拿到第一次的效果进行第2次RPC 调用
     rpcInvoking_2()...........
​
    // 拿到第2次的效果进行第三次RPC 调用、
     rpcInvoking_3()...........
​
    // 拿到第三次的效果进行第四次RPC 调用、
     rpcInvoking_4()...........
​
    // 组装效果回来
​
    return BasicInfo;
   }

是的，这个数据的获取就是这么的扯淡。。。假如运用循环的办法，当ids数据量在500个左右的时分，这个接口回来的时刻再8s左右，这是万万不能接受的，那假如ids数据更多呢？所以不能用for循环去遍历ids呀，这样确实是太费时了。

已然长途调用避免不了，那就想办法让这个接口快一点，这时分就想到了多线程去处理，然后就想到运用CompletableFuture异步调用：

CompletableFuture多线程异步调用

      List<BasicInfo> basicInfoList = new ArrayList<>();
   CompletableFuture<List<BasicInfo>> future = CompletableFuture.supplyAsync(() -> {
      ids.forEach(id -> {
        getBasicData(basicInfoList, id);
       });
      return basicInfoList;
    });
    try {
      List<BasicInfo> basicInfos = future.get();
    } catch (Exception e) {
      e.printStackTrace();
    } 

这儿补偿一点：CompletableFuture是否运用默许线程池的根据，和机器的CPU中心数有关。当CPU中心数减1大于1时，才会运用默许的线程池（ForkJoinPool），否则将会为每个CompletableFuture的任务创建一个新线程去实行。即，CompletableFuture的默许线程池，只要在双核以上的机器内才会运用。在双核及以下的机器中，会为每个任务创建一个新线程，等于没有运用线程池，且有资源耗尽的危险。

默许线程池，池内的中心线程数，也为机器中心数减1，这儿我们的机器是8核的，也就是会创建7个线程去实行。

上面这种办法尽管结束了多线程异步实行，但是假如ids调集许多话，仍然会很慢，由于future.get();也是阻塞的，必须等候一切的线程实行结束才干回来效果。

改进CompletableFuture多线程异步调用

想让速度更快一点，就想到了把ids进行分隔：

 int pageSize = ids.size() > 8 ? ids.size() >> 3 : 1;
 List<List<Long>> partitionAssetsIdList = Lists.partition(ids, pageSize);

由于我们CPU核数为8核，一得当ids的大小小于8时，就敞开8个线程，每个线程分一个。这儿的>>3(右移运算)相当于ids的大小除以2的3次方也就是除以8；右移运算符比较除功率会高。究竟现在是在优化提高速度。

假如这儿的ids的大小是500个，就是敞开9个线程，其间8个线程是处理62个数据，另一个线程处理4个数据，由于有余数会另开一个线程处理。具体代码如下：

        int pageSize = ids.size() > 8 ? ids.size() >> 3 : 1;
    List<List<Long>> partitionIdList = Lists.partition(ids, pageSize);
    List<CompletableFuture<?>> futures = new ArrayList<>();
    //假如ids为500，这儿会分隔成9份，也就是partitionIdList.size()=9；遍历9次,也相当于创建了9个CompletableFuture方针，前8个CompletableFuture方针处理62个数据。第9个处理4个数据。
    partitionIdList.forEach(partitionIds -> {
      List<BasicInfo> basicInfoList = new ArrayList<>();
      CompletableFuture<List<BasicInfo>> future = CompletableFuture.supplyAsync(() -> {
        partitionIds.forEach(id -> {
          getBasicData(basicInfoList, id);
         });
        return basicInfoList;
       });
      futures.add(future);
     });
    // 把一切线程实行的效果进行汇总
    List<BasicInfo> basicInfoResult = new ArrayList<>();
    for (CompletableFuture<?> future : futures) {
      try {
        basicInfoResult.addAll((List<BasicInfo>)future.get());
       } catch (Exception e) {
        e.printStackTrace();
       }
     }

假如ids的大小等于500，就会被分隔成9份，创建9个CompletableFuture方针，前8个CompletableFuture方针处理62个数据(id),第9个处理4个数据(id)。这62个数据又会被分红7个线程去实行(CPU核数减1个线程)。经过分隔之后充分利用了CPU。速度也从8s减到1-2s。得到了总监和同事的夸奖，一起也被写到正向事情中；哈哈哈哈。

在出产环境中遇到的坑。

上面说了那么多还没有说到坑在哪里，下面我们就说说坑在哪里?

本地和检验都没有啥问题，那就找个时刻上出产呗，升级到出产环境，发现这个接口阻塞了，超时了。。。

刚被记录到正向事情，可不想在被记录个负向时刻。感觉去看日志。

发现日志就实行了将ids进行分隔，后边循环去创建CompletableFuture方针之后的代码都没有在实行了。然后我第一感觉检验是future.get()获取效果的时分阻塞了，所以一贯没有效果回来。

排查问题进程

我们要处理这个问题就要看看问题出现在哪里？

当实行到这个接口时分我们第一时刻看了看CPU的运用率：

这是访问接口之前：

发现实行这个接口时PID为10348的这个进程的CPU忽然的高了起来。

紧接着运用jps -l :打印出我们服务进程的PID

PID为10348正式我们现在实行这个服务。

接着我就详细的看一下这个PID为10348的进程下哪里线程占用的高：

发现这几个占用的相对高一点：

紧接着运用jstack指令生成java虚拟机其时时刻的线程快照，生成线程快照的首要目的是定位线程出现长时间间断的原因，如线程间死锁、死循环、恳求外部资源导致的长时间等候等。 线程出现间断的时分经过jstack来查看各个线程的调用仓库，就可以知道没有响应的线程到底在后台做什么事情，或许等候什么资源

jstack -l 10348 >/tmp/10348.log，运用此指令将PID为10348的进程下一切线程快照输出到log文件中。

一起我们将线程比较的PID转换成16进制：printf “%x\n” 10411

我们将转换成16进制的数值28ab，28a9在10348.log中查找一下：

看到线程的快照发现这不是本次批改的接口呀。看到日志4处但是也是用了CompletableFuture。找到对应4处的代码发现这是监听mq音讯，然后异步去实行，代码类型这样：

经过查看日志发现这个mq音讯处理很频频，每秒都会有许多的数据上来。

我们知道CompletableFuture默许是运用ForkJoinPool作为线程池。莫非mq运用ForkJoinPool和我其时接口运用的都是同一个线程池中的线程？莫非是共用的吗？

MQ监听运用的线程池：

我们其时接口运用的线程池：

它们运用的都是ForkJoinPool.commonPool()公共线程池中的线程！

看到这个定论就很好理解了，我们现在批改的接口运用的线程池中的线程全部都被MQ音讯处理占用，我们批改优化的接口得不到资源，所以一贯处于等候。

一起我们在线程快照10348.log日志中也看到我们优化的接口对应的线程处于WAITING状况！

这儿- parking to wait for <0x00000000fe2081d8>必定也是MQ消费线程中的某一个。由于MQ消费音讯比较多，每秒都会监听到许多的数据，线程的快照日志收集不全。所以在10348.log中没有找到，这不影响我们批改bug。问题的原因现已找到了。

处理问题

上面我们知道两边运用的都是公共静态线程池，我们只要让他们各用各的就行了：自定义一个线程池：ForkJoinPool pool = new ForkJoinPool();

        int pageSize = ids.size() > 8 ? ids.size() >> 3 : 1;
    List<List<Long>> partitionIdList = Lists.partition(ids, pageSize);
    List<CompletableFuture<?>> futures = new ArrayList<>();
    partitionIdList.forEach(partitionIds -> {
      List<BasicInfo> basicInfoList = new ArrayList<>();
        //从头创建一个ForkJoinPool方针就可以了
      ForkJoinPool pool = new ForkJoinPool();
      CompletableFuture<List<BasicInfo>> future = CompletableFuture.supplyAsync(() -> {
        partitionIds.forEach(id -> {
          getMonitoringCoverage(basicInfoList, id);
         });
        return basicInfoList;
      //在这儿运用
       },pool);
      futures.add(future);
     });
    // 把一切线程实行的效果进行汇总
    List<BasicInfo> basicInfoResult = new ArrayList<>();
    for (CompletableFuture<?> future : futures) {
      try {
        basicInfoResult.addAll((List<BasicInfo>)future.get());
       } catch (Exception e) {
        e.printStackTrace();
       }
     }

这样他们就各自用各自的线程池中的线程了。不会存在资源的等候现场了。

总结：

之所以检验环境和开发环境没有出现这样的问题是由于这两个环境mq没有监听到音讯。许多的音讯都在出产环境中才会出现。由于检验环境的数据量达不到出产环境的数据量，所以有些问题在检验环境领会不出来。

码字不易，多多支撑。仍是那句话：不积跬步,无以致千里.不积小流,无以成江海！