RESTful API in Spring 上通量行为的解释

Question

我正在努力理解 Spring 的 Webflux 和响应式 API，希望能解释一下当使用注释 vice a [=41 声明 REST 端点时我看到的通量行为差异=].

我看到的是通量（流？）在通过注释定义的 REST 端点获取时正在执行，而不是从使用 route/handler 语义定义的 REST 端点获取。

谁能解释观察到的行为的差异？下面提供的代码和控制台输出...

注意：DemoRequestHandler.getOddIntsMult() 中的注释代码将导致 flux 执行并迭代包含的整数以 find/return 奇数值。我想我真正的问题是 "why is the subscribe() required in one instance and not the other?"

带注释的 REST 控制器...

import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;

import static org.springframework.web.bind.annotation.RequestMethod.GET;

@RestController
public class DemoFluxController {

    @RequestMapping(method=GET, value="/v1/fluxMult")
    public Flux<Integer> getMultFlux() {
        System.out.println("DEBUG -> FluxController.getMultFlux()");
        return DemoFlux.getOddInts(DemoFlux.multIntFlux);
    }
}

测试 class 用于 return 具有奇数整数的通量...

import reactor.core.publisher.Flux;

public class DemoFlux {

    public static Flux<Integer> multIntFlux = Flux.range(1, 20);

    private static boolean isOdd(Integer intVal) {
        System.out.printf("DEBUG -> DemoFlux.isOdd( %d )%n", intVal);
        return intVal % 2 != 0;
    }

    public static Flux<Integer> getOddInts(Flux<Integer> intFlux) {
        return intFlux.filter(DemoFlux::isOdd);
    }
}

声明备用 REST 端点的路由器实现...

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.MediaType;
import org.springframework.web.reactive.function.server.RequestPredicates;
import org.springframework.web.reactive.function.server.RouterFunction;
import org.springframework.web.reactive.function.server.RouterFunctions;
import org.springframework.web.reactive.function.server.ServerResponse;

@Configuration
public class DemoRouter {

    @Bean
    public RouterFunction<ServerResponse> route(DemoRequestHandler requestHandler) {
        return RouterFunctions.route(RequestPredicates.GET("/v2/fluxMult")
                          .and(RequestPredicates.accept(MediaType.APPLICATION_JSON)), requestHandler::getOddIntsMult);
    }
}

与 REST 端点的路由关联的请求处理程序。

import org.springframework.http.MediaType;
import org.springframework.stereotype.Component;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.server.ServerRequest;
import org.springframework.web.reactive.function.server.ServerResponse;
import reactor.core.publisher.Mono;

@Component
public class DemoRequestHandler {

    public Mono<ServerResponse> getOddIntsMult(ServerRequest request) {
        System.out.println("DEBYG -> DemoRequestHandler.getOddIntsMult()");

        return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON)
            .body(BodyInserters.fromObject(DemoFlux.getOddInts(DemoFlux.multIntFlux)));
//                .body(BodyInserters.fromObject(DemoFlux.getOddInts(DemoFlux.multIntFlux).subscribe()));
    }
}

来自运行的控制台输出。请注意不同的行为取决于访问哪个 REST 端点...

DEBUG -> DemoFluxController.getMultFlux()
DEBUG -> DemoFlux.getOddInts()
DEBUG -> DemoFlux.isOdd( 1 )
DEBUG -> DemoFlux.isOdd( 2 )
DEBUG -> DemoFlux.isOdd( 3 )
DEBUG -> DemoFlux.isOdd( 4 )
DEBUG -> DemoFlux.isOdd( 5 )
DEBUG -> DemoFlux.isOdd( 6 )
DEBUG -> DemoFlux.isOdd( 7 )
DEBUG -> DemoFlux.isOdd( 8 )
DEBUG -> DemoFlux.isOdd( 9 )
DEBUG -> DemoFlux.isOdd( 10 )
DEBUG -> DemoFlux.isOdd( 11 )
DEBUG -> DemoFlux.isOdd( 12 )
DEBUG -> DemoFlux.isOdd( 13 )
DEBUG -> DemoFlux.isOdd( 14 )
DEBUG -> DemoFlux.isOdd( 15 )
DEBUG -> DemoFlux.isOdd( 16 )
DEBUG -> DemoFlux.isOdd( 17 )
DEBUG -> DemoFlux.isOdd( 18 )
DEBUG -> DemoFlux.isOdd( 19 )
DEBUG -> DemoFlux.isOdd( 20 )

DEBUG -> DemoRequestHandler.getOddIntsMult()
DEBUG -> DemoFlux.getOddInts()

Answer 1

它们的行为不同，因为您没有对它们进行相同的编码。

人们在使用 webflux/reactive 编程时最常犯的错误之一是他们在应用程序中间订阅，因为他们想要或需要某些东西的具体价值。

WebFlux/reactive 编程基本上就是声明我们希望在有人订阅时发生的一系列事件。在这种情况下，它是客户端订阅的时间。所以你能感觉到有什么不对劲真是太好了，因为确实有！

你的问题在这里：

public Mono<ServerResponse> getOddIntsMult(ServerRequest request) {
    System.out.println("DEBYG -> DemoRequestHandler.getOddIntsMult()");

    return ServerResponse.ok()
        .contentType(MediaType.APPLICATION_JSON)
        .body(BodyInserters.fromObject(              // This body inserter here
            DemoFlux.getOddInts(DemoFlux.multIntFlux)));
}

实际情况是 BodyInserter 需要一个具体值，如果您查看它的源代码，您会发现主体插入器将主体包裹在单声道中。因此，您所做的是将 Flux 包装在 Mono 中，因此您实际上得到的是 Mono<Flux<Integer>>.

在你使用 webflux 工作了一段时间之后，当你得到类似这样的东西时，你就会知道是这种情况：

{
    "disposed": true,
    "scanAvailable": true
}

然后你知道你得到的是一个包装的 mono/flux 而你错过了某个地方的平面图。

如果您重写它并删除 BodyInserter，它将按预期工作。

public Mono<ServerResponse> getOddIntsMult(ServerRequest request) {
    System.out.println("DEBYG -> DemoRequestHandler.getOddIntsMult()");

    return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON)
            .body(DemoFlux.getOddInts(DemoFlux.multIntFlux), Integer.class);                
}

所以没有区别，你的 @RequestMapping(method=GET, value="/v1/fluxMult") 不使用 DemoRequestHandler 因此你得到不同的结果。

RESTful API in Spring 上通量行为的解释

Explanation of flux behavior on RESTful API in Spring

rest

spring-boot

spring-webflux