为什么从WebClient访问RESTendpoint的处理程序会被访问两次？

Question

这是第二次尝试，通过修改的演示代码，希望能更好地说明这个问题。代码已经被剥离，除去了所有的元素，除了那些演示正在遇到的问题的元素。

添加注意：一些额外的测试做了，结果张贴作为一个答案（副扩展这篇文章）。这可能是“预期的行为”，但我仍在努力理解“为什么”。

代码“起作用”，因为它返回预期的信息（字符串或字符串列表）。但是，当使用WebClient访问返回流量的RESTendpoint(localhost:8080/test/democlient)时，会对关联的处理程序（DemoMainHandler.getAll（））进行两次调用。我不知道在DemoMainHandler.getAll（）上的第二个调用是在哪里进行的，但我担心如果在生产环境中发生这种情况，可能会出现性能问题。

在所提供的代码中，所有内容都在单个Spring Webflux应用程序下运行，因此DemoClient代码没有单独的进程。

但是，访问localhost:8080/test/democlient上的RESTendpoint会产生一些相关的结果。通过Flux返回给Postman的字符串值看起来还可以，但是

1. 访问RESTendpoint时调用的DemoclientHandler.getAll()
2. *DemoclientHandler.getAll（）调用Democlient.getAll()
3. democlient.getAll（）使用WebClient访问localhost:8080/test/demomain
上的RESTendpoint
4. democlient.getAll（）使用flatMapMany迭代返回的ClientResponse并从响应体提取流量
5. 由Democlient.getAll（）生成的通量返回给DemoclientHandler.getAll()
6. DemoclientHandler.getAll（）检查流量，确定它有一个或多个元素，并将流量以ServerResponse的形式返回给初始客户端（在本例中为邮递员）
7. 邮递员然后打开通量（map？flatmap？）并显示返回的字符串（“CallMeOnce”）

我不理解的是，为什么DemoclientHandler.getAll（）会被第二次调用，如控制台的第二个system.out.println（）输出所示。它似乎与使用通量作为返回类型有关？

控制台输出

2019-10-07 08:16:18.953  INFO 9384 --- [           main] c.example.testdupe.TestDupeApplication   : Starting TestDupeApplication on M7730-LFR with PID 9384 (D:\sandbox\TestDupe\build\classes\java\main started by LesR in D:\sandbox\TestDupe)
2019-10-07 08:16:18.953  INFO 9384 --- [           main] c.example.testdupe.TestDupeApplication   : No active profile set, falling back to default profiles: default
2019-10-07 08:16:20.062  INFO 9384 --- [           main] o.s.b.web.embedded.netty.NettyWebServer  : Netty started on port(s): 8080
2019-10-07 08:16:20.062  INFO 9384 --- [           main] c.example.testdupe.TestDupeApplication   : Started TestDupeApplication in 1.324 seconds (JVM running for 1.871)

***** Invoke localhost:8080/test/DemoClient/{id}
DemoClientHandler.getById( ServerRequest )
DemoClient.getById( 2 )
DemoMainHandler.getById( ServerRequest )

***** Invoke localhost:8080/test/DemoClient
DemoClientHandler.getAll( ServerRequest )
DemoClientHandler.getAll() >> BEFORE invoking demoClient.getAll()
DemoClient.getAll()
DemoClient.getAll() >> RETURN fluxString
DemoClientHandler.getAll() >>  AFTER invoking demoClient.getAll()
DemoMainHandler.getAll( ServerRequest )
DemoMainHandler.getAll( ServerRequest )

示例代码

@SpringBootApplication
public class TestDupeApplication {

    public static void main(String[] args) {
        SpringApplication.run(TestDupeApplication.class, args);
    }
}


@Configuration
public class DemoClientRouter {

    @Bean
    public RouterFunction<ServerResponse> clientRoutes(DemoClientHandler requestHandler) {
        return nest(path("/test"),
                nest(accept(APPLICATION_JSON),
                        RouterFunctions.route(RequestPredicates.GET("/DemoClient"), requestHandler::getAll)
                                    .andRoute(Requesthtml" target="_blank">Predicates.GET("/DemoClient/{id}"), requestHandler::getById)));
    }
}


@Component
public class DemoClientHandler {

    @Autowired
    DemoClient demoClient;

    public Mono<ServerResponse> getAll(ServerRequest request) {
        System.out.println("DemoClientHandler.getAll( ServerRequest )");
        System.out.println("DemoClientHandler.getAll() >> BEFORE invoking demoClient.getAll()");
        Flux<String> fluxString = demoClient.getAll();
        System.out.println("DemoClientHandler.getAll() >>  AFTER invoking demoClient.getAll()");

        return fluxString.hasElements().flatMap(hasElement -> {
            return hasElement ? ServerResponse.ok()
                                              .contentType(MediaType.APPLICATION_JSON)
                                              .body(fluxString, String.class)
                              : ServerResponse.noContent().build();
        });
    }

    public Mono<ServerResponse> getById(ServerRequest request) {
        System.out.println("DemoClientHandler.getById( ServerRequest )");
        Mono<String> monoString;

        return demoClient.getById( 2 ).flatMap(stringVal -> ServerResponse.ok()
                                                                          .contentType(MediaType.APPLICATION_JSON)
                                                                          .body(Mono.just(stringVal), String.class))
                                      .switchIfEmpty(ServerResponse.notFound().build());
    }
}


@Component
public class DemoClient {

    private final WebClient client;

    public DemoClient() {
        client = WebClient.create();
    }

    public Flux<String> getAll() {
        System.out.println("DemoClient.getAll()");
        Flux<String> fluxString;

        fluxString = client.get().uri("http://localhost:8080/test/DemoMain")
                                 .accept(MediaType.APPLICATION_JSON)
                                 .exchange()
                                 .flatMapMany(response -> response.bodyToFlux(String.class));

        // fluxString = client.get().uri("http://localhost:8080/test/DemoMain")
        //                          .accept(MediaType.APPLICATION_JSON)
        //                          .retrieve()
        //                          .bodyToFlux(String.class);

        System.out.println("DemoClient.getAll() >> RETURN fluxString");
        return fluxString;
    }

    public Mono<String> getById(int id) {
        System.out.printf("DemoClient.getById( %d )%n", id);
        return client.get().uri("http://localhost:8080/test/DemoMain/" + id)
                           .accept(MediaType.APPLICATION_JSON)
                           .exchange()
                           .flatMap(response -> response.bodyToMono(String.class));
    }
}


@Configuration
public class DemoMainRouter {
    @Bean
    public RouterFunction<ServerResponse> demoPOJORoute(DemoMainHandler requestHandler) {
        return nest(path("/test"),
                nest(accept(APPLICATION_JSON),
                        RouterFunctions.route(RequestPredicates.GET("/DemoMain"), requestHandler::getAll)
                                    .andRoute(RequestPredicates.GET("/DemoMain/{id}"), requestHandler::getById)));
    }
}


@Component
public class DemoMainHandler {

    public Mono<ServerResponse> getAll(ServerRequest request) {
        System.out.println("DemoMainHandler.getAll( ServerRequest )");

        return ServerResponse.ok()
                             .contentType(MediaType.APPLICATION_JSON)
                             .body(Flux.just("Call", "Me", "Once"), String.class);
    }

    public Mono<ServerResponse> getById(ServerRequest request) {
        System.out.println("DemoMainHandler.getById( ServerRequest )");

        return ServerResponse.ok()
                             .contentType(MediaType.APPLICATION_JSON)
                             .body(Mono.just("Only One"), String.class);
    }
}

添加此代码是为了支持后续讨论...

@Component
public class DemoClient {

    private final WebClient client;

    public DemoClient() {
        client = WebClient.create();
    }

    public Flux<String> getAll() {
        Flux<String> fluxString;

        Mono<ClientResponse> monoCR = client.get().uri("http://localhost:8080/test/DemoMain")
                                                  .accept(MediaType.APPLICATION_JSON)
                                                  .exchange();

        fluxString = monoCR.flatMapMany(clientResponse -> clientResponse.bodyToFlux(String.class));

//        fluxString.subscribe();
//        return fluxString;
        return Flux.just("Foo", "Bar");
    }

Answer 1

后续讨论。不是一个真正的答案，但感觉它的方向是正确的。

修改了democlient.gatall（）来“解链”通量/流上的操作，希望获得一些洞察力。以下是我做的/发现的：

1. 我引入了一个单变量来保存localhost上的WebClient访问结果:8080/test/demomain
2. 我独立地调用了monocr.flatMapMany（）以获得返回的通量，并将该通量分配给FluxString
3. 我添加了一个FluxString.Subscribe；语句，只是为了能够订阅返回的通量而无需对其执行任何操作
4. 我介绍了返回flux.just（“foo”，“bar”）；语句，如果我选择不返回FluxString

当我注释掉fluxString.subscribe()'并返回fluxString时；语句，则DemoMainHandler.getAll（）没有输出。我想这“并不奇怪”，因为生成的Flux上没有订阅任何东西，所以没有调用DemoMainHandler.getAll()，因为不需要Flux。

当我取消对fluxString.subscribe（）的注释时；但保留返回FluxString；注释后，我看到了DemoMainHandler.getAll（）的一个println（）输出。再一次，我认为这是“不是一个意外”，因为通量现在是订阅的，即使没有做任何结果。因此，DemoMainHandler.getAll（）被调用并输出其println（）内容。

最后，我注释掉了fluxString.subscribe（）；并返回flux.just（“foo”,“bar”）；语句，以及未注释的*return fluxString；“。这会从DemoMainHandler.getAll（）中产生我一直询问的两个println（）输出。

基于订阅返回的Flux的结果，我假设DemoMainHandler.getAll（）输出的第一个println（）是代表Postman（即“最终消费者”）隐式订阅的结果。但是，这仍然给我留下了一个问题：“为什么DemoMainHandler.getAll（）的第二个println（）输出？”Reactor真的在订阅时调用DemoMainHandler.getAll（）一次，在处理实际内容时再调用一次吗？还是？