在 Java 中将 [chained] CompletableFutures 写入 CSV

Question

我有一个 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, CompletableFuture<HashMap<String, String>>> 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 I 运行的结果，然后对它们调用 allOf，但这会导致奇怪的行为，即 Solr 的属性停止显示上去后几百项。数据库中的属性确实显示了每个项目，这让我觉得第一个 supplyAsync 总是有效，但随后的 thenApply 将属性添加到 HashMap<String, CompletableFuture<HashMap<String, String>>> itemsToAttributesMapping; 提供的原始 HashMap<String, CompletableFuture<HashMap<String, String>>> itemsToAttributesMapping; =30=]停止工作。

任何关于可能是什么问题的见解都将不胜感激。也许我对 CompletableFuture 的理解不正确，尤其是在链接和解析期货方面？也许这是超时问题，或者线程正在丢失？我扩展的最后一种方法表明问题可能是 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）揭示了什么？真的有你说的那么快吗？

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

在 Java 中将 [chained] CompletableFutures 写入 CSV

Writing [chained] CompletableFutures to CSV in Java

java

asynchronous

solr

hashmap

completable-future