用 Java 将 [chained] CompletableFutures 写入 CSV

编程问答 2022-06-02

问题描述

我有一个 HashMap<String,CompletableFuture<HashMap<String,String>>> 将一个项目映射到它的属性，例如{ "pizza" -> { "calories" -> "120","fat" -> "12" } }，其中的属性是从不同的数据源检索的。

例如，我们从数据库中获取 "fat" 属性，而从 Solr 中获取 "calories" 属性。

当我最初从数据库中检索 "fat" 时，为了不阻塞主线程，我使用了 supplyAsync，例如：

  public CompletableFuture<HashMap<String,String>> getFat(String name,boolean) {
    return CompletableFuture.supplyAsync(new supplier<HashMap<String,String>>() {
      @Override
      public HashMap<String,String> get() {
        HashMap<String,String> attributes = new HashMap<>();
        
        ... do work ...
      
        attributes.put(name,attributes);
        return attributes;
      }
   })
 }

然后我将它与对 Solr 的异步调用链接起来，这样我最终就会有一个异步 Hashmap 将项映射到它们的属性，即 HashMap<String,String>>> itemsToAttributesMapping;（所以我循环遍历哈希图的键并使用新属性，使用 thenApply 访问旧属性）。

我通过将数据映射到 csv 来完成，这就是问题出现的地方：

       File file = new File(home + "/Downloads/rmsSkuValidationResults.csv");

       try{
          FileWriter outputfile = new FileWriter(file);
          CSVWriter writer = new CSVWriter(outputfile);

            for(String itemKey : itemsToAttributesMapping.keySet()) {
                itemsToAttributesMapping.get(itemKey).thenAccept(attributes -> {

                String[] row = { attributes.get("calories"),attributes.get("fat")
                        
                        ... more attributes ...

                        };
                writer.writeNext(row);
                });
            }

         writer.close();
      }
      catch(Exception e){
        e.printstacktrace();
      }

按原样打印到 CSV 文件可以正常处理大约 800-1100 个项目，但在此之后停止写入并且程序终止。

我尝试了上述的变体，包括使用 get 而不是 thenAccept，或者在 join 之后添加 thenAccept 导致程序挂起（异步计算很快,不应该挂）。

我还尝试存储我运行的 thenAccepts 的结果，然后对它们调用 allOf，但这会导致奇怪的行为，即 Solr 的属性在几百个项目后停止显示.数据库中的属性确实出现在每个项目中，这让我认为第一个 supplyAsync 总是有效，但后续 thenApply 将属性添加到 HashMap<String,String>>> itemsToAttributesMapping; 提供的原始 supplyAsnc {1}} 停止工作。

对可能是什么问题的任何见解将不胜感激。也许我对 CompletableFuture 的理解是不正确的，尤其是在链接和解决期货方面？也许这是一个超时问题，或者线程正在丢失？我扩展的最后一个方法表明问题可能出在 thenApplys?

解决方法

以下是您上面代码的粗略说明，正如您所拥有的：

get(itemKey1) then at some unspecified time in the future writeNext(attr1)
get(itemKey2) then at some unspecified time in the future writeNext(attr2)
get(itemKey3) then at some unspecified time in the future writeNext(attr3)
get(itemKey4) then at some unspecified time in the future writeNext(attr4)
get(itemKey5) then at some unspecified time in the future writeNext(attr5)
get(itemKey6) then at some unspecified time in the future writeNext(attr6)
get(itemKey7) then at some unspecified time in the future writeNext(attr7)
attr1 finally delivered; writeNext(attr1)
get(itemKey8) then at some unspecified time in the future writeNext(attr8)
attr2 finally delivered; writeNext(attr2)
attr3 finally delivered; writeNext(attr3)
get(itemKey9) then at some unspecified time in the future writeNext(attr9)
no more items; writer.close()
attr4 finally delivered; oops,writer closed
attr5 finally delivered; oops,writer closed
attr6 finally delivered; oops,writer closed
attr7 finally delivered; oops,writer closed
attr8 finally delivered; oops,writer closed
attr9 finally delivered; oops,writer closed

您提到您尝试过 .get() 和 .join()。这基本上会使程序同步，但这是一个很好的调试步骤。它会将执行更改为：

get(itemKey1) then at some unspecified time in the future writeNext(attr1)
attr1 finally delivered; writeNext(attr1)
get(itemKey2) then at some unspecified time in the future writeNext(attr2)
attr2 finally delivered; writeNext(attr2)
get(itemKey3) then at some unspecified time in the future writeNext(attr3)
attr3 finally delivered; writeNext(attr3)
...
...
...
get(itemKey9) then at some unspecified time in the future writeNext(attr9)
attr9 finally delivered; writeNext(attr9)
no more items; writer.close()

这应该有效。将输出添加到您的每个阶段（您未显示的 thenApply 以及 thenAccept）显示了什么？真的有你说的那么快吗？

请显示更多代码。尤其是链接部分，如果这是您认为可能存在问题的地方。

asynchronouscompletable-futurehashmaphashmapjavajavasolrsolrsolr