用Java将[chained]CompletableFutures写入CSV

Question

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

例如，我们从数据库中获得“fat”属性，而从Solr中获得“calories”属性。

当我最初从DB检索“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 >itemstoattributesmapping； （因此我循环遍历Hashmap的键，并用新属性更新值，使用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，或者在thenaccept后面添加join导致程序挂起（异步计算很快，不应该挂起）。

我还尝试存储我运行的thenaccepts的结果，然后对其调用allof，但这会导致奇怪的行为，在几百个项目之后，Solr的属性就不再显示了。数据库中的属性确实会为每个项显示出来，这使我认为第一个SupplyAsync总是工作的，但是后续的ThenApply将属性添加到HashMap >ItemstoAttributesMapping； 提供的SupplyAsNC中的ThenApply停止工作。

如果你能洞悉问题所在，我将不胜感激。也许我对CompletableFuture的理解是不正确的，尤其是在链接和解决Futures上？这可能是超时问题，或者线程丢失了？我所阐述的最后一种方法暗示问题可能是thenapplys？

Answer 1

下面是上面的代码正在做什么的粗略说明：

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)会显示什么？真的像你说的那么快吗？

请显示更多代码。尤其是链接部分，如果你认为这是问题所在的话。