1

我实际上对组装时间和订阅时间感到困惑。我知道单声道是懒惰的，并且在订阅之前不会被执行。下面是一个方法。

    public Mono<UserbaseEntityResponse> getUserbaseDetailsForEntityId(String id) {
        GroupRequest request = ImmutableGroupRequest
                .builder()
                .cloudId(id)
                .build();

        Mono<List<GroupResponse>> response = ussClient.getGroups(request);
        List<UserbaseEntityResponse.GroupPrincipal> groups = new CopyOnWriteArrayList<>();


        response.flatMapIterable(elem -> elem)
                .toIterable().iterator().forEachRemaining(
                    groupResponse -> {
                        groupResponse.getResources().iterator().forEachRemaining(
                            resource -> {
                                groups.add(ImmutableGroupPrincipal
                                        .builder()
                                        .groupId(resource.getId())
                                        .name(resource.getDisplayName())
                                        .addAllUsers(convertMemebersToUsers(resource))
                                        .build());
                            }
                        );
                    }
                );
        log.debug("Response Object - " + groups.toString());
        ImmutableUserbaseEntityResponse res = ImmutableUserbaseEntityResponse
                .builder()
                .userbaseId(id)
                .addAllGroups(groups)
                .build();


        Flux<UserbaseEntityResponse.GroupPrincipal> f = Flux.fromIterable(res.getGroups())
        .parallel()
                .runOn(Schedulers.parallel())
                .doOnNext(groupPrincipal -> getResourcesForGroup((ImmutableGroupPrincipal)groupPrincipal, res.getUserbaseId()))
                .sequential();

        return Mono.just(res);
    }

这会在Mono<List<GroupResponse>> response = ussClient.getGroups(request);不调用订阅的情况下执行，但是除非我调用订阅，否则下面不会执行。

Flux<UserbaseEntityResponse.GroupPrincipal> f = Flux.fromIterable(res.getGroups())
        .parallel()
                .runOn(Schedulers.parallel())
                .doOnNext(groupPrincipal -> getResourcesForGroup((ImmutableGroupPrincipal)groupPrincipal, res.getUserbaseId()))
                .sequential();

我可以在组装时间与订阅方面获得更多信息吗？

4

1 回答 1

2

“在您订阅之前什么都不会发生”并非在所有情况下都是正确的。Mono发布者（或Flux）将在三种情况下执行：

  • 您订阅；
  • 你阻止；
  • 出版商很“热”。

请注意，上述场景都适用于整个反应链 - 即，如果我订阅发布者，则上游的所有内容（依赖于该发布者）也会执行。这就是为什么框架可以并且应该在需要时调用 subscribe，从而导致控制器中定义的反应链执行。

在您的情况下，它实际上是其中的第二个-您正在阻塞，这本质上是“订阅并等待结果”。通常阻塞的方法会被清楚地标记，但情况并非总是如此 - 在您的情况下，它是执行阻塞的toIterable()方法：Flux

将此 Flux 转换为对Iterator.next()调用的惰性 Iterable 阻塞。

但是啊，你说，我没有打电话Iterator.next()——什么给了？！

好吧，隐含地你是通过调用forEachRemaining()：

默认实现的行为如下：

 while (hasNext())
     action.accept(next());

...并且根据上述规则，由于ussClient.getGroups(request)位于此阻塞调用的上游，因此它会被执行。

于 2020-07-03T22:40:45.107 回答