Flux 不会在“then”之前等待元素完成

Question

我无法理解这个问题，我不确定我做错了什么

我想等待flux结束，然后返回serverResponse的mono

我已附上代码 sn-p，doOnNext 将填充 categoryIdToPrintRepository

我查看了如何在通量结束后返回单声道，发现“then”但仍然在处理 onNextSite 之前执行“then”方法，这导致错误：

java.lang.IllegalArgumentException: 'producer' type is unknown to ReactiveAdapterRegistry

我做错了什么？

 public Mono<ServerResponse> retrieveCatalog(ServerRequest ignored) {
        return Mono.just("start").flatMap(id ->
                Flux.fromIterable(appSettings.getSites())
                        .subscribeOn(ForkJoinPoolScheduler.create("SiteCatalogScheduler"))
                        .doOnNext(this::onNextSite)
                        .then(Mono.from(ServerResponse.ok().body(categoryIdToPrintRepository.getSortedTreeValues(), String.class))));

    }

    private void onNextSite(Integer siteId) {
        IntStream.range(1, appSettings.getCatalogMaxValue()).parallel().forEach(catalogId -> {
            Optional<SiteCatalogCategoryDTO> cacheData =
                    siteCatalogCacheUseCaseService.getSiteCatalogResponseFromCache(siteId, catalogId);
            cacheData.ifPresentOrElse(siteCatalogCategoryDTO -> {/*do nothing already exist in cache*/},
                    () -> {
                    Mono<SiteCatalogCategoryDTO> catalogCategoryDTOMono = WebClient.create(getUri(siteId, catalogId))
                            .get().retrieve().bodyToMono(SiteCatalogCategoryDTO.class);
                    catalogCategoryDTOMono.subscribe(siteCatalogCategoryDTO ->
                            handleSiteServerResponse(siteCatalogCategoryDTO, siteId, catalogId));
            });
        });
    }


    private void handleSiteServerResponse(SiteCatalogCategoryDTO siteCatalogCategoryDTO,
                                          int siteId, int catalogId) {
        if (siteCatalogCategoryDTO.getResponseStatus().equals(ResponseStatus.SUCCESS))
            Flux.fromIterable(siteCatalogCategoryDTO.getMappingList())
                    .subscribe(mapSCC -> {
                        categoryIdToPrintRepository.insertIntoTree(mapSCC.getCategoryId(),
                                "Site " + siteId + " - Catalog " + catalogId + " is mapped to category " + "\"" +
                                        mapSCC.getCategoryName() + "\" (" + mapSCC.getCategoryId() + ")");
                        siteCatalogCacheUseCaseService.insertIntoSiteCatalogCache(siteId, catalogId, siteCatalogCategoryDTO);
                    });
    }

Answer 1

你做错了几件事，你不应该在你的应用程序中subscribe，并且你有 void 方法，除非在特定的地方，否则不应该在反应式编程中使用。

这里是一些示例代码：

// Nothing will happen, we are not returning anything, we can't subscribe
private void doSomething() {
    Mono.just("Foo");
}

// complier error
doSomething().subscribe( ... );

您的应用程序是调用客户端的publisher，是订阅者，这就是为什么我们向调用客户端返回 Mono 或 Flux，它们是 subscribe。

你已经这样解决了：

private void doSomething() {
    Mono.just("Foo").subscribe( ... );
}

doSomething();

现在您正在订阅自己以使事情运行，这不是正确的方式，如前所述，调用客户端是订阅者，而不是您。

正确方法：

private Mono<String> doSomething() {
    return Mono.just("Foo");
}

// This is returned out to the calling client, they subscribe
return doSomething();

当 Mono/Flux 完成时，它会发出一个信号，这个信号将触发链中的下一个和下一个。

所以我对你需要做的事情的看法如下：

• 删除all subscribes，如果你想做一些事情，有flatmap，map，doOnSuccess等函数。保持链一直保持不变客户。
• 删除所有 void 函数，确保它们返回 Flux 或 Mono，如果你想不返回一些东西，使用返回 Mono<Void> Mono.empty() 函数，这样链就完成了。

一旦您使用 Mono/Flux，您需要处理返回，以便其他人可以链接。

更新：

为了触发then，你必须返回一些东西，它会在之前的单声道/通量完成时返回。

示例：

private Flux<String> doSomething() {
    return Flux.just("Foo", "Bar", "FooBar")
               .doOnNext(string -> {
                   return // return something
               });
}

// Ignore what was return from doSomething and return something else when the flux has completed (so only trigger on the completed signal from the flux)
return doSomething().then( ... );