使用 CompletableFuture 的并发数据库调用答案

【问题标题】：concurrent DB calls using CompletableFuture使用 CompletableFuture 的并发数据库调用
【发布时间】：2020-07-18 17:03:41
【问题描述】：

我正在尝试提高我的应用程序的仪表板页面的性能。这是一个 Spring Boot 应用程序，休眠用于 Dao 层。我需要针对 5 种不同的警报类型（基本上是过滤器）获取参与者的数量。

到目前为止，我正在尝试同时启动数据库查询：

  Map<Long,CompletableFuture<Long>> furtureMap = new HashMap<>();
      Map<Long, Long> alertMap = new HashMap<>();
      if(CollectionUtils.isNotEmpty(alertList)) {

        for(AlertMasterI18N alertMasterI18N : alertList) {
          dashboardFilterDto.setAlertId(alertMasterI18N.getAlertMaster().getId());
          setDatesForDashboardFilterDto(dashboardFilterDto);
          CompletableFuture<Long> future = CompletableFuture.supplyAsync(() -> {
            try {
              return dashboardDao.getParticipantsCount(dashboardFilterDto);
            } catch (DaoException e) {
              e.printStackTrace();
            }
            return 0L;
          });
          furtureMap.put(alertMasterI18N.getAlertMaster().getId(), future);
        }
       /* CompletableFuture.allOf(furtureMap.values().toArray(new CompletableFuture[furtureMap.size()])).get(); */
       furtureMap.entrySet().stream().forEach( entry -> {
         entry.getValue().thenAccept( count -> alertMap.put(entry.getKey(),count));
       });
      }

不知何故，我得到的结果比按顺序执行得到的结果更奇怪。所有警报都会返回相同的计数，有时它是 0，而如果我恢复代码，它会给出正确的结果。我也尝试过使用 CompletableFuture.allOf 和 callig get() 但那时所有警报的计数始终为 0。

这是我第一次尝试异步编程谁能告诉我这里有什么遗漏的吗？

【问题讨论】：

你能分享这个代码dashboardDao.getParticipantsCount(dashboardFilterDto) 吗？

标签： java spring multithreading concurrency completable-future

【解决方案1】：

假设 DAO 层一切正常，我能够在提供的代码中看到的唯一缺失点是最后调用 get() 或 join()。如果没有这个调用，CompletableFuture 将不会真正执行。

下面的大纲在最后一个 forEach 循环中添加所有新创建的CompletableFuture，然后等待它们。

    Map<Long,CompletableFuture<Long>> furtureMap = new HashMap<>();
    Map<Long, Long> alertMap = new HashMap<>();
    if(CollectionUtils.isNotEmpty(alertList)) {

        for (AlertMasterI18N alertMasterI18N : alertList) {
            dashboardFilterDto.setAlertId(alertMasterI18N.getAlertMaster().getId());
            setDatesForDashboardFilterDto(dashboardFilterDto);
            CompletableFuture<Long> future = CompletableFuture.supplyAsync(() -> {
                try {
                    return dashboardDao.getParticipantsCount(dashboardFilterDto);
                } catch (DaoException e) {
                    e.printStackTrace();
                }
                return 0L;
            });
            furtureMap.put(alertMasterI18N.getAlertMaster().getId(), future);
        }
        List<CompletableFuture<Void>> finalCollector = new ArrayList<>();
        /* CompletableFuture.allOf(furtureMap.values().toArray(new CompletableFuture[furtureMap.size()])).get(); */
        furtureMap.entrySet().stream().forEach(entry -> {
            finalCollector.add(entry.getValue()
                    .thenAccept(count -> alertMap.put(entry.getKey(), count)));
        });
        /* Wait for all the results */
        CompletableFuture
                .allOf(finalCollector.toArray(new CompletableFuture[finalCollector.size()]))
                .join();
    }

如果这不起作用，则很可能是 DAO 层中出现了问题，导致异步执行失败（同时成功执行顺序执行）。分享您在 DAO 层中获取数据的方式将进一步阐明问题的根源。

【讨论】：