Spring Webflux带依赖多顺序Web请求的实现优化与问题咨询

原始实现代码 #

public Mono<List<ParcelInfo>> parcelEventsForReference(String parcelId, Postcode postcode) {
    return connector1.getParcelReference(parcelId) // 1. Mono<ResolvedParcelIds> (Web请求1)
        .flatMapIterable(ResolvedParcelIds::getParcelIds) // 2. Flux<ResolvedParcelId>
        .flatMap(id -> parcelEventsFor(id) // 3. Mono<ParcelInfo> 
            .map(parcel -> redactIfNecessary(parcel, id, postcode))) // 4. Flux<ParcelInfo>
        .collectList() // 5. Mono<List<ParcelInfo>>
        .flatMapIterable(list -> list) // 6. Flux<ParcelInfo>
        .flatMap(this::getTrackingEvents)
        .collectList()
        .doOnError(throwable -> log.error(EXTERNAL_SERVICE_ERROR, throwable.getMessage()));
 }

private Mono<ParcelInfo> parcelEventsFor(ResolvedParcelId id) {
    return connector2.getParcelBy(id).zipWith(connector3.getAdditionalInstructions(uid)) // (Web请求2,3)
            .map(tuple -> combineParcelEvents(tuple.getT1(), tuple.getT2()));
 }

private Mono<ParcelInfo> getTrackingEvents(ParcelInfo parcel) {
    TrackingRequest request = mapToRequest(parcel);
    return connector4.getTrackingForParcel(request) // (Web请求4)
        .map(trackingResponse -> addTracking(parcel, trackingResponse));
}

private ParcelInfo addTracking(ParcelInfo parcel, TrackingResponse trackingResponse) {
    parcelProgress.setTracking(trackingResponse);
    return parcel;
 }

public class ParcelInfo { 
    private List<ParcelBase> parcel;
    private TrackingResponse trackingResponse;
    private String additionalInstructions;
}

public class ResolvedParcelIds {
    private List<ResolvedParcelId> parcelIds;
}

疑问解答 #

1. 是否应使用zipWhen运算符确保第4次请求仅在第1、2、3次请求完成后执行？若应使用，该如何实现？ #

你的当前实现已经保证了第4次请求（getTrackingEvents）是在第1、2、3次请求完成后才执行的——因为flatMap只会在上游流的元素产生后才触发后续操作，而上游的flatMap(id -> parcelEventsFor(id))已经涵盖了请求1、2、3的完整流程。

zipWhen并不适合这个场景：它的核心作用是将当前Mono的结果作为参数触发另一个Mono，再合并两者结果，更适合一对一的依赖场景。而你的需求是每个ParcelInfo都需要单独触发请求4，用flatMap才是更贴合的选择。

此外，当前代码里的两次collectList+flatMapIterable属于冗余操作，可以优化流程减少不必要的聚合与展开：

public Mono<List<ParcelInfo>> parcelEventsForReference(String parcelId, Postcode postcode) {
    return connector1.getParcelReference(parcelId) // 请求1
            .flatMapIterable(ResolvedParcelIds::getParcelIds)
            .flatMap(id -> parcelEventsFor(id) // 并行执行请求2、3
                    .map(parcel -> redactIfNecessary(parcel, id, postcode)))
            .flatMap(this::getTrackingEvents) // 每个ParcelInfo处理完成后触发请求4
            .collectList()
            .doOnError(throwable -> log.error(EXTERNAL_SERVICE_ERROR, throwable.getMessage()));
}

优化后的逻辑更简洁，同时依然严格保证请求4在1、2、3次请求完成后执行。

2. 若第1、2、3次请求未返回HTTP 200成功响应，如何阻止第4次请求执行？ #

首先要明确：doOnError只是错误的副作用处理（比如打日志），不会改变流的状态。要阻止请求4执行，需要让前序请求的错误正确传播，中断整个流：

• 请求1失败：connector1.getParcelReference(parcelId)会返回错误Mono，整个流直接进入doOnError，后续所有操作（包括请求2、3、4）都不会触发。
• 请求2或3失败：parcelEventsFor(id)会返回错误Mono，默认情况下flatMap会终止当前元素的处理，但继续处理其他ResolvedParcelId对应的请求。如果需要只要有一个请求2/3失败就终止整个流、完全阻止请求4执行，可以添加onErrorStop()：

.flatMap(id -> parcelEventsFor(id)
        .map(parcel -> redactIfNecessary(parcel, id, postcode))
        .onErrorStop()) // 单个元素处理出错时，终止整个Flux

如果希望单个请求2/3失败不影响其他元素的处理，但出错的元素不触发请求4，可以用onErrorReturn返回默认值或空对象：

.flatMap(id -> parcelEventsFor(id)
        .map(parcel -> redactIfNecessary(parcel, id, postcode))
        .onErrorReturn(new ParcelInfo())) // 出错时返回默认值，跳过该元素的请求4

注意：如果你的连接器使用了onErrorResume之类的操作符将错误转为默认结果，流会继续执行，此时请求4依然会被触发——这种情况下需要根据业务需求调整连接器的错误处理逻辑，确保失败场景下错误能正确传播。

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