Java 单声道重复调用，直到收集的结果竞争答案

【问题标题】：Java mono repeat call until collected results competeJava 单声道重复调用，直到收集的结果竞争
【发布时间】：2020-10-13 15:03:45
【问题描述】：

从 C# 迁移过来后，我正在学习 Java/Reactor。我精通 C# async-await 方法来处理非阻塞调用，并且正在努力适应 Flux/Mono。

我正在实现一个解决方案，我需要通过 Java SDK 调用 ElasticSearch，获取结果，应用额外的过滤器来去除 ES 结果，并通过 ES 继续分页，直到我的最终结果收集完成。

ES SDK 不支持 Reactor，但有一些 Java 适配器代码示例，它接受 ES 回调并转换为单声道（我在这里看到与 C# async-await 直接相关，因为这是一个非阻塞调用到 ES）。然后我要努力的是下一点 - 我需要从 ES 单声道中获取结果，过滤它们。

我通过调用其他外部服务来根据 ES 调用的结果获取额外数据来做到这一点，因此我需要知道 ES 单声道结果的每个内容页面的 id，然后才能应用过滤（有效一种块），然后应用内存过滤器，如果我没有足够的内容，则返回 ES 以获取下一页...重复直到我有足够的数据或 ES 没有更多结果.

与 C# 相比，这似乎很难实现，但我可能只是没有正确理解 Java 范例。

我的问题是我不能使用“block()”，因为这会在 Reactor 3.2 中引发错误，所以我真的不知道如何“等待”直到对 ES 和外部服务的单声道调用完成，直到继续.在 C# 中，这就像调用带有等待处理隐式回调的 Async 方法一样简单

我的阻塞版本（在 IntellJ 中工作，通过 maven 发布然后在网络服务器中运行时失败）是有效的：

do {
 var sr = GetSearchRequest(xxxx);
 this.elasticsearch.results(sr)
   .map(r -> chunk.addAdd(r))
   .block();

 if (chunk.size() == 0 {
   isComplete = true;
 }
 else {
  var filtered = postFilterResults(chunk);
  finalResults.add(filtered);
  if (finalResults.size() = MAXIMUM_RESULTS) {
    isComplete = true;
  }
  esPage = esPage + 1;
while (isComplete == false);

如果我尝试 subscribe() 或其他非阻塞的 reaktor 调用，那么（显然）代码会跳过“get ES”位并点击 do-while，重复循环直到 ES 的回调最终发生并且订阅的地图被调用。

我认为我需要为每个 ES 调用执行一个“异步块”，但我不知道如何。

【问题讨论】：

标签： java reactor

【解决方案1】：

回答我自己的问题... IMO 的根本问题是 Flux/Mono 根本不像任何现有的编程风格，因为它绝对迫使您在反应堆要求的流畅风格内工作。这与 C# Linq 非常相似，但它几乎是一个“假朋友”，因为即使是循环之类的东西也需要在 Reactor 中。

在这种情况下，要解决的关键问题是分页并在循环中继续执行此操作。非常不清楚如何通过订阅通量“锁定”原始参数来实现这一点，因此重复订阅调用只会再次获得相同的页面。解决方案是使用 Flux.defer 方法，该方法强制在每次重复调用时延迟构建订阅。然后，您需要原子整数来跟踪不同调用中的页面计数器。同样，这是 C# 为您处理的东西，因此它可以吸引 .net 开发人员。

类似：

//The response from the elasticsearch adapter is a Flux<T> but we do not want to filter
//results on a row by row basis as this incurs one call for each row to the DB/Network
//(as appropriate). We choose to batch these up

var result = new SearchResult();
var page = new AtomicInteger();
var chunkSize = new AtomicInteger();

//Use a defer so we recalculate the subscription to the search with the new page count
var results = Flux.defer(() -> elasticsearch.results(GetSearchRequest(request, lc, pf, page.get()))
       .doOnComplete(() -> {
           chunkSize.set(0);
           page.getAndAdd(1);
       })
       .collectList()
       .map(chunk -> {
           chunkSize.set(chunk.size());
           return chunk;
       })
       .map(chunk -> postFilterResults(request, chunk, pf))
       .map(filtered -> result.getDocuments().addAll(filtered)));


//Repeat the deferred flux (recalculating each time) until we have enough content or we don't get anything from the search engine
return results
        .repeat()
        .takeUntil(r -> chunkSize.get() == 0 || result.getDocuments().size() >= this.elasticsearch.getMaximumSearchResults())
        .take(this.elasticsearch.getMaximumSearchResults())
        .collectList()
        .flatMap(r -> {
            result.setTotalHits(result.getDocuments().size());
            return Mono.just(result);
        });

【讨论】：