Spring WebFlux响应式编程如何逐个发送列表内对象？

你需要先把Mono<List<BalanceAlertAttribute>>转换为逐个发射列表元素的Flux，再逐个执行转换、发送逻辑，修改后的代码如下：

public Mono<AlertServiceResponse> sendBalanceAlertDataToPrepaid(BalanceAlertServiceRequest cartasRequest) {
    return Mono.just(cartasRequest.getBalanceAlertServiceRequestData().getBalanceAlertList())
        // 1. 将Mono包裹的List转为逐个发射元素的Flux
        .flatMapMany(Flux::fromIterable)
        // 2. 执行单个元素的转换逻辑，不要用doOnEach，doOnEach是副作用操作符不参与流元素转换
        .map(this::mapBalanceAlertServiceRequestToBalanceChangeAlertResponse)
        // 3. 逐个调用发送接口，如果需要严格按列表顺序发送可以把flatMap换成concatMap
        .flatMap(prepaidClient::sendBalanceAlertData)
        // 4. 等待所有元素发送完成后，返回统一的响应
        .then(Mono.just(cartaResponse(cartasRequest.getServiceRequestId())));
}

关键逻辑说明 #

• 关于doOnEach的使用疑问：

doOnEach属于副作用操作符，仅用于监听流的信号（元素发射、完成、报错）做日志、埋点这类不修改流内容的操作，不会将转换后的结果传递到下游，所以不能用来做业务转换逻辑，元素转换直接用map即可。

• 发送模式选择：
  • 用flatMap：发送请求是异步并发执行，性能更高，不保证和原列表顺序一致
  • 用concatMap：严格按照原列表顺序发送，前一个发送完成才会发起下一个请求，性能相对低
• 可选异常处理：如果需要单个元素发送失败不影响其他元素，可以在flatMap中添加降级逻辑：

  .flatMap(item -> prepaidClient.sendBalanceAlertData(item)
      .onErrorResume(e -> {
          log.error("发送余额告警失败, item:{}", item, e);
          // 错误后返回空信号，不中断整个流
          return Mono.empty();
      }))
  

内容的提问来源于stack exchange，提问作者lukederienzo