如何合并两个并行执行两个反应器通量，返回列表并合并结果

Question

我有一个LegacyAcCountDto，我需要从两个不同的来源建立一个列表。一个是本地JPA存储库，另一个是Web服务调用。Web服务版本具有JPA数据源不可用的帐户状态。我需要并行执行两个调用，当它们都完成时，我需要找到Web服务列表的legacyId，并用从Web服务中提取的帐户状态填充列表。整个想法是返回一个包含完整DTO的列表。我不需要把它保存回网络服务或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;
}

merge语句中的每个函数都返回一个LegacyDTO流

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
    })

我不确定接下来调用什么函数，以便能够访问两个列表并从第二次调用中获取帐户状态，并且能够合并它，并且不返回并行通量类型，而不仅仅是LegacyDto的通量

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);

不建议在反应流中使用阻塞数据库调用。如果你有选择的话，你可以添加R2DBC驱动程序，让整个过程都是被动的。