java - 如何合并两个并行执行两个 react 器通量，返回列表并合并结果

Question

我有一个 LegacyAccountDto，我需要从两个不同的来源构建一个列表。一个是本地 JPA 存储库，另一个是 Web 服务调用。 Web 服务版本具有可用的 accountStatus，而 JPA 数据源没有。我需要作为 Fluxes 并行执行两个调用，然后当它们都完成时，我需要找到 Web 服务列表的 legacyId 并使用从 Web 服务中提取的 accountStatus 填充列表。整个想法是返回一个包含已完成 DTO 的列表。我不需要将其保存回 Web 服务或 JPA 存储库

DTO:

@Data
@JsonInclude(Include.NON_NULL)
public class LegacyAccountDto {
  private UUID id;
  private UUID organizationId;
  private String name;
  private String website;
  private Long legacyAccountId;
  private LocalDateTime legacyCreated;
  private String accountType;
  private String accountState;
}

合并语句中的每个函数都返回 LegacyDTO 的 Flux

Flux<LegacyAccountDto> completed = Flux.merge(
      getLegacyAccountsFromSvc(accountNames),
      Flux.fromIterable(accountMapper.accountListToLegacyAccountDtoList(accountRepository.getAccountsByNames(accountNames).get()))
    )
    .parallel()
    .runOn(Schedulers.elastic())
    .???????((list1, list2) -> {
      list2.map(l2 -> {
        //find list1 by legacyId
        //set l2.accountStatus = l1.accountstatus
      })
      //return the completed list as a flux
    })

我不确定接下来要调用什么函数才能访问两个列表并从第二次调用中获取 accountStatus 并能够合并它并且让它不返回并行通量类型而不仅仅是通量旧版Dto

Answer 1

你可以这样做:

Mono<List<LegacyAccountDto>> firstMonoList = getLegacyAccountsFromSvc(accountNames).collectList();
Mono<List<EntityX> secondMonoList = accountMapper.accountListToLegacyAccountDtoList(accountRepository.getAccountsByNames(accountNames).get()).collectList();

Mono.zip(firstMonoList, secondMonoList)
.map(listTuple -> {
  // do the searching and map to list of dtos
  return resultingList;
})
.flatMapMany(Flux::fromIterable);

不建议在 react 流中使用阻塞数据库调用。如果您可以选择添加 R2DBC驱动程序并使整个事情 react 。