并行呼叫流量

Question

我想用一个参数调用一个函数，它并行返回一个Mono，并以列表或通量的形式返回结果。

我正在调用一个函数，它返回一个mono

 return Flux.fromIterable(txnReqList)
                .parallel(5)
                .runOn(Schedulers.parallel())
                .flatMap(id -> getResponse(id))
                .sequential().collectList();

响应是Mono

请帮助同步进行，以便通话能在1秒内自动进行。

@Autowired
RSocketRequester requester;

@Override
public Mono < TransactionResponseDto > getResponse(TransactionRequestDto request) {
    //log.info(request.toString());
    return this.requester.route("post.transaction")
        .data(request).retrieveMono(TransactionResponseDto.class).log();
}

@Override
public Mono < List < TransactionResponseDto >> getAllResponse(TransactionRequestDto request) {
    TransactionRequestDto txnReqOrig1 = new TransactionRequestDto();
    txnReqOrig1.setExecOrder(1);
    txnReqOrig1.setProvider("PROV0");
    txnReqOrig1.setAcntId("0");
    txnReqOrig1.setTxnAmt(100.00);
    TransactionRequestDto txnReqOrig2 = new TransactionRequestDto();
    txnReqOrig1.setExecOrder(2);
    txnReqOrig2.setProvider("PROV1");
    txnReqOrig2.setAcntId("1");
    txnReqOrig2.setTxnAmt(101.00);
    TransactionRequestDto txnReqOrig3 = request;
    txnReqOrig3.setExecOrder(3);
    List < TransactionRequestDto > txnReqList = new ArrayList < TransactionRequestDto > ();
    txnReqList.add(txnReqOrig3);
    txnReqList.add(txnReqOrig1);
    txnReqList.add(txnReqOrig2);

    return Flux.fromIterable(txnReqList)
        .parallel(5)
        .runOn(Schedulers.parallel())
        .flatMap(id - > getResponse(id))
        .sequential().collectList();

}

提前谢谢




Answer 1

这里有一个简单的例子，演示了通量产生的元素的并行处理。

Mono<List<Integer>> listMono = Flux.range(1, 5) <-- produces 5 events
            .parallel(5) <-- runs all 5 in parallel
            .runOn(Schedulers.parallel())
            .doOnNext(id -> slowTask()) <-- sleeps for x seconds
            .doOnNext(System.out::println) <-- prints out the id
            .sequential()
            .collectList();

StepVerifier.create(listMono.then())
            .verifyComplete();

当运行这段代码时，您将看到执行的持续时间是x，其中x是慢任务（）所花费的秒数。如果这不是并行的，而是连续的，执行将需要5倍的时间。

请告诉我您是否可以运行此代码段，以及您是否看到与上述相同的行为？

Answer 2

你需要回答的最重要的问题是getResponse做什么。基于它的名字，以及你用线程睡眠来模拟它的行为，我假设它做了一些阻塞IO操作。并行调度器并不是为了这个目的，它是为了CPU目的的操作。由于这个原因，它有尽可能多的CPU核可用的线程（或有时乘以一些较低的数字）。

阻塞IO操作应卸载到弹性调度程序，如下所示：

return Flux.fromIterable(txnReqList)
           .flatMap(id -> Mono.defer(() -> getResponse(id)).subscribeOn(Schedulers.elastic()))
           .collectList();

弹性线程池能够根据负载进行“伸缩”。

然而，如果您可以在getResponse方法中使用一些非阻塞IO库（Spring WebClient、R2DBC等），则是最被动的方法。在这种情况下，您根本不需要担心调度器和线程，您将拥有一个高度可扩展的解决方案。

另外，在被动世界中，你可以使用Mono模拟延迟。延迟。如果有非阻塞IO，可以使用Mono。延迟用于测试目的。如果你有阻塞IO，那么线程。睡眠是正确的方法。