如何在 Spring WebFlux 的反应链中异步运行方法？

我正在尝试在基于 project reactor 的应用程序中的现有反应链中异步执行方法。 doupdatelayoutinasync 方法旨在执行繁重的后台任务，但我的方法似乎没有按预期工作。这是我当前的实现：

public Mono publishPackage(String branchedPackageId) {
    PackagePublishingMetaDTO publishingMetaDTO = new PackagePublishingMetaDTO();
    publishingMetaDTO.setPublishEvent(true);

    return packageRepository
            .findById(branchedPackageId, packagePermission.getPublishPermission())
            .switchIfEmpty(Mono.error(new AppsmithException(
                    AppsmithError.ACL_NO_RESOURCE_FOUND, FieldName.PACKAGE_ID, branchedPackageId)))
            .flatMap(originalPackage -> {
                String nextVersion = PackageUtils.getNextVersion(originalPackage.getVersion());

                Package packageToBePublished = constructPackageToBePublished(originalPackage);

                originalPackage.setVersion(nextVersion);
                originalPackage.setLastPublishedAt(packageToBePublished.getLastPublishedAt());
                publishingMetaDTO.setOriginPackageId(branchedPackageId);
                publishingMetaDTO.setWorkspaceId(originalPackage.getWorkspaceId());

                Mono unsetCurrentLatestMono = packageRepository.unsetLatestPackageByOriginId(originalPackage.getId(), null);
                Mono saveOriginalPackage = packageRepository.save(originalPackage);
                Mono savePackageToBePublished = packageRepository.save(packageToBePublished);

                return unsetCurrentLatestMono
                        .then(Mono.zip(saveOriginalPackage, savePackageToBePublished))
                        .flatMap(tuple2 -> {
                            Package publishedPackage = tuple2.getT2();
                            publishingMetaDTO.setPublishedPackage(publishedPackage);

                            return modulePackagePublishableService
                                    .publishEntities(publishingMetaDTO)
                                    .flatMap(publishedModules -> {
                                        if (publishedModules.isEmpty()) {
                                            return Mono.error(new AppsmithException(
                                                    AppsmithError.PACKAGE_CANNOT_BE_PUBLISHED,
                                                    originalPackage.getUnpublishedPackage().getName()));
                                        }
                                        return moduleInstancePackagePublishableService
                                                .publishEntities(publishingMetaDTO)
                                                .then(Mono.defer(() ->
                                                        newActionPackagePublishableService.publishEntities(publishingMetaDTO))
                                                        .then(Mono.defer(() ->
                                                                actionCollectionPackagePublishableService
                                                                        .publishEntities(publishingMetaDTO))));
                                    })
                                    .then(Mono.defer(() -> autoUpgradeService.handleAutoUpgrade(publishingMetaDTO)));
                        })
                        .as(transactionalOperator::transactional)
                        .then(Mono.defer(() -> doUpdateLayoutInAsync(publishingMetaDTO)));
            });
}

private Mono doUpdateLayoutInAsync(PackagePublishingMetaDTO publishingMetaDTO) {
    Mono> updateLayoutsMono = Flux.fromIterable(publishingMetaDTO.getAutoUpgradedPageIds())
            .flatMap(pageId -> updateLayoutService
                    .updatePageLayoutsByPageId(pageId)
                    .onErrorResume(throwable -> {
                        log.warn("Update layout failed for pageId: {} with error: {}", pageId, throwable.getMessage());
                        return Mono.just(pageId);
                    }))
            .collectList();

    // Running the updateLayoutsMono task asynchronously
    updateLayoutsMono.subscribeOn(Schedulers.boundedElastic()).subscribe();

    return Mono.just(Boolean.TRUE);
}

问题： 我希望 doupdatelayoutinasync 在后台运行，而反应链的其余部分完成。然而，该方法似乎是同步执行的，反应链并没有按预期继续。

Civitai

AI艺术分享平台！海量SD资源和开源模型。

下载

问题： 如何确保 doupdatelayoutinasync 异步运行并且不会阻塞反应链的继续？