如何组合两个或多个参数化的 Monos Webflux 最佳实践

Posted 2023-02-27

【中文标题】如何组合两个或多个参数化的 Monos Webflux 最佳实践

【英文标题】：How to Combine Two or more parametrised Monos Webflux Best practice

【发布时间】：2021-09-13 00:47:12

【问题描述】：

所以我调用了一个服务“ServiceA”，它通过 webclient 进行外部调用，返回一个单声道， 然后我想调用另一个服务“serviceB”，它也进行类似的外部调用，这次传递一些数据 “serviceA”单声道，也许还有其他一些外部参数。

Mono<ResponseA> responseAmono = serviceA.getdataA("some parameter");
Mono<ResponseB> responseBmono = serviceB.getdataB("data from responseAmono", "some parameter")

目前我有这个实现。

responseAmono
        .doOnSuccess(data -> 
            if (data.field == true) 
                serviceB.getdataB(data, "some parameter");
            
        )
        .subscribe(x -> , e -> sout("serviceA Subscription Error! "+ e.getMessage()));

但是我遇到了线程池的挑战，由于PoolAcquirePendingLimitException，我不得不从最多 1000 个连接增加 pendingAcquireQueue。

那么，结合这两个单声道以获得良好性能（即>4000 tps）的最佳策略是什么？

在调用内部单声道“serviceB”时分离关注点。

注意。我的订阅也是在serviceB上实现的。

【参考方案1】：

doOnSuccess() 应该只为同步副作用调用，而不是响应操作。如果您实际上是在 doOnSuccess() 调用中阻塞，这可能就是您遇到线程问题的原因。

相反，flatMap() 将是一个更好的选择，因为它可以保持一切反应。如果您仍然想返回服务 A 的结果而忽略服务 B 的结果，您可以类似以下操作：

responseAmono
    .flatMap(data -> 
        if (data.field == true) 
            return serviceB.getdataB(data, "some parameter").thenReturn(data);
        
        else 
            return Mono.just(data);
        
    )

【讨论】：

非常感谢@Michael Berry，让我重构并做一些基准测试。会回应的。