异步编程 - 11 Spring WebFlux的异步非阻塞处理

public interface HttpHandler {
    /**
     * Handle the given request and write to the response.
     * @param request current request
     * @param response current response
     * @return indicates completion of request handling
     */
    Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response);

}

HttpHandler handler = ...
ReactorHttpHandlerAdapter adapter = new ReactorHttpHandlerAdapter(handler);
HttpServer.create(host, port).newHandler(adapter).block();

HttpHandler handler = ...
Servlet servlet = new TomcatHttpHandlerAdapter(handler);

Tomcat server = new Tomcat();
File base = new File(System.getProperty("java.io.tmpdir"));
Context rootContext = server.addContext("", base.getAbsolutePath());
Tomcat.addServlet(rootContext, "main", servlet);
rootContext.addServletMappingDecoded("/", "main");
server.setHost(host);
server.setPort(port);
server.start();

@RestController
public class PersonHandler {

    @GetMapping("/getPerson")
    Mono<String> getPerson() {
        return Mono.just("jiaduo");
    }
}

@RestController
public class PersonHandler {

    @GetMapping("/getPersonList")
    Flux<String> getPersonList() {
        return Flux.just("jiaduo", "zhailuxu", "guoheng");
    }
}

@RestController
public class PersonHandler {

    @GetMapping("/getPersonList")
    Flux<String> getPersonList() {
        return Flux.just("jiaduo", "zhailuxu", "guoheng").map(e -> {
            System.out.println(Thread.currentThread().getName());
            return e;
        });
    }

reactor-http-nio-2
reactor-http-nio-2
reactor-http-nio-2

@RestController
public class PersonHandler {
    @GetMapping("/getPersonList")
    Flux<String> getPersonList() {
        return Flux.just("jiaduo", "zhailuxu", "guoheng")
                .publishOn(Schedulers.elastic())//1.1 切换到异步线程执行
                .map(e -> {//1.2打印调用线程
            System.out.println(Thread.currentThread().getName());
            return e;
        });
    }

elastic-2
elastic-2
elastic-2

@RestController
public class PersonHandler {
    // 1.0创建线程池
    private static final ThreadPoolExecutor bizPoolExecutor = new ThreadPoolExecutor(8, 8, 1, TimeUnit.MINUTES,
            new LinkedBlockingQueue<>(10));
    
    @GetMapping("/getPersonList")
    Flux<String> getPersonList() {
        return Flux.just("jiaduo", "zhailuxu", "guoheng")
                .publishOn(Schedulers.fromExecutor(bizPoolExecutor))//1.1 切换到异步线程执行
                .map(e -> {//1.2打印调用线程
            System.out.println(Thread.currentThread().getName());
            return e;
        });
    }

@Configuration
public class FunctionModelConfig {
    
    @Bean
    public FunctionPersonHandler  handler() {
        return new FunctionPersonHandler();
    }
    @Bean
    public RouterFunction<ServerResponse>routerFunction(final FunctionPersonHandler handler) {
        RouterFunction<ServerResponse> route =  RouterFunctions.route()//1
                .GET("/getPersonF",RequestPredicates.accept(MediaType.APPLICATION_JSON), handler::getPerson)//2
                .GET("/getPersonListF",RequestPredicates.accept(MediaType.APPLICATION_JSON), handler::getPersonList)//3
                .build();//4

        return route;
    }
}

public class FunctionPersonHandler {
    // 1.0创建线程池
    private static final ThreadPoolExecutor bizPoolExecutor = new ThreadPoolExecutor(8, 8, 1, TimeUnit.MINUTES,
            new LinkedBlockingQueue<>(10));

    Mono<ServerResponse> getPersonList(ServerRequest request) {
        // 1.根据request查找person列表
        Flux<String> personList = Flux.just("jiaduo", "zhailuxu", "guoheng")
                .publishOn(Schedulers.fromExecutor(bizPoolExecutor))// 1.1 切换到异步线程执行
                .map(e -> {// 1.2打印调用线程
                    System.out.println(Thread.currentThread().getName());
                    return e;
                });
        
        // 2.返回查找结果
        return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON).body(personList, String.class);
    }

    Mono<ServerResponse> getPerson(ServerRequest request) {
        // 1.根据request查找person,
        Mono<String> person = Mono.just("jiaduo");
        // 2.返回查找结果
        return ServerResponse.ok().contentType(MediaType.APPLICATION_JSON).body(person, String.class);
    }
}

pool-2-thread-1
pool-2-thread-2
pool-2-thread-2

<dependencyManagement> 
    <dependencies>
        <dependency>
            <groupId>io.projectreactor</groupId>
            <artifactId>reactor-bom</artifactId>
            <version>Californium-RELEASE</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

<dependencies>
    <dependency>
        <groupId>io.projectreactor.netty</groupId>
        <artifactId>reactor-netty</artifactId>
    </dependency>
</dependencies>

import reactor.core.publisher.Mono;
import reactor.netty.DisposableServer;
import reactor.netty.http.server.HttpServer;

public class ReactorNetty {
    public static void main(String[] args) {
        DisposableServer server = HttpServer.create()//1.创建http服务器
                .host("localhost")//2.设置host
                .port(8080)//3.设置监听端口
                .route(routes -> routes//4.设置路由规则
                        .get("/hello", (request, response) -> response.sendString(Mono.just("Hello World!")))
                        .post("/echo", (request, response) -> response.send(request.receive().retain()))
                        .get("/path/{param}",
                                (request, response) -> response.sendString(Mono.just(request.param("param"))))
                        .ws("/ws", (wsInbound, wsOutbound) -> wsOutbound.send(wsInbound.receive().retain())))
                .bindNow();

        server.onDispose().block();//5.阻塞方式启动服务器，同步等待服务停止
    }
}

protected ConfigurableApplicationContext createApplicationContext() {
    Class<?> contextClass = this.applicationContextClass;
    if (contextClass == null) {
        try {
            //a 环境类型
            switch (this.webApplicationType) {
            case SERVLET://a.1 Web servlet环境
                contextClass = Class.forName(DEFAULT_SERVLET_WEB_CONTEXT_CLASS);
                break;
            case REACTIVE://a.2 Web Reactive环境
                contextClass = Class.forName(DEFAULT_REACTIVE_WEB_CONTEXT_CLASS);
                break;
            default://a.3 非Web环境
                contextClass = Class.forName(DEFAULT_CONTEXT_CLASS);
            }
        }
        catch (ClassNotFoundException ex) {
            throw new IllegalStateException(
                    "Unable create a default ApplicationContext, "
                            + "please specify an ApplicationContextClass",
                    ex);
        }
    }
    return (ConfigurableApplicationContext) BeanUtils.instantiateClass(contextClass);
}

//默认非Web环境时
public static final String DEFAULT_CONTEXT_CLASS = "org.springframework.context."
        + "annotation.AnnotationConfigApplicationContext";

//web Servlet环境时默认的上下文
public static final String DEFAULT_SERVLET_WEB_CONTEXT_CLASS = "org.springframework.boot."
        + "web.servlet.context.AnnotationConfigServletWebServerApplicationContext";

//反应式Web环境时默认的上下文
public static final String DEFAULT_REACTIVE_WEB_CONTEXT_CLASS = "org.springframework."
        + "boot.web.reactive.context.AnnotationConfigReactiveWebServerApplicationContext";

public SpringApplication(ResourceLoader resourceLoader, Class<?>... primarySources) {
    ...
    this.webApplicationType = WebApplicationType.deduceFromClasspath();
    ...
}
下面我们看WebApplicationType的deduceFromClasspath方法：

static WebApplicationType deduceFromClasspath() {
    //b.判断是不是REACTIVE类型
    if (ClassUtils.isPresent(WEBFLUX_INDICATOR_CLASS, null)
            && !ClassUtils.isPresent(WEBMVC_INDICATOR_CLASS, null)
            && !ClassUtils.isPresent(JERSEY_INDICATOR_CLASS, null)) {
        return WebApplicationType.REACTIVE;
    }
    //c.判断是不是非Web类型
    for (String className : SERVLET_INDICATOR_CLASSES) {
        if (!ClassUtils.isPresent(className, null)) {
            return WebApplicationType.NONE;
        }
    }

    //SERVLET环境
    return WebApplicationType.SERVLET;
}

//spring mvc 分派器
private static final String WEBMVC_INDICATOR_CLASS = "org.springframework."
        + " web.servlet.DispatcherServlet";
// reactive web分派器
private static final String WEBFLUX_INDICATOR_CLASS = "org."
        + "springframework.web.reactive.DispatcherHandler";
//Jersey Web 项目容器类
private static final String JERSEY_INDICATOR_CLASS = "org.glassfish.jersey.servlet.ServletContainer";
//Servlet容器所需要的类
private static final String[] SERVLET_INDICATOR_CLASSES = { "javax.servlet.Servlet",
            "org.springframework.web.context.ConfigurableWebApplicationContext" };

protected ReactiveWebServerFactory getWebServerFactory() {
    //d 从bean工厂中获取所有ReactiveWebServerFactory类型的Bean实例的名字
    String[] beanNames = getBeanFactory()
            .getBeanNamesForType(ReactiveWebServerFactory.class);
    //e 不存在则抛出异常
    if (beanNames.length == 0) {
        throw new ApplicationContextException(
                "Unable to start ReactiveWebApplicationContext due to missing "
                        + "ReactiveWebServerFactory bean.");
    }
    if (beanNames.length > 1) {
        throw new ApplicationContextException(
                "Unable to start ReactiveWebApplicationContext due to multiple "
                        + "ReactiveWebServerFactory beans : "
                        + StringUtils.arrayToCommaDelimitedString(beanNames));
    }
    //f 存在则获取第一个实例
    return getBeanFactory().getBean(beanNames[0], ReactiveWebServerFactory.class);
}

class ReactiveWebServerFactoryConfiguration {

    //f.1将NettyReactiveWebServerFactory注入容器
    @Configuration
    @ConditionalOnMissingBean(ReactiveWebServerFactory.class)
    @ConditionalOnClass({ HttpServer.class })
    static class EmbeddedNetty {

        @Bean
        public NettyReactiveWebServerFactory nettyReactiveWebServerFactory() {
            return new NettyReactiveWebServerFactory();
        }

    }
    //f.2注入TomcatReactiveWebServerFactory实例
    @Configuration
    @ConditionalOnMissingBean(ReactiveWebServerFactory.class)
    @ConditionalOnClass({ org.apache.catalina.startup.Tomcat.class })
    static class EmbeddedTomcat {

        @Bean
        public TomcatReactiveWebServerFactory tomcatReactiveWebServerFactory() {
            return new TomcatReactiveWebServerFactory();
        }

    }
    //f.3注入JettyReactiveWebServerFactory实例
    @Configuration
    @ConditionalOnMissingBean(ReactiveWebServerFactory.class)
    @ConditionalOnClass({ org.eclipse.jetty.server.Server.class })
    static class EmbeddedJetty {

        @Bean
        public JettyReactiveWebServerFactory jettyReactiveWebServerFactory() {
            return new JettyReactiveWebServerFactory();
        }

    }
    //f.4注入UndertowReactiveWebServerFactory实例
    @ConditionalOnMissingBean(ReactiveWebServerFactory.class)
    @ConditionalOnClass({ Undertow.class })
    static class EmbeddedUndertow {

        @Bean
        public UndertowReactiveWebServerFactory undertowReactiveWebServerFactory() {
            return new UndertowReactiveWebServerFactory();
        }
    }
}

public static ServerManager get(ReactiveWebServerFactory factory) {
        return new ServerManager(factory);
    }

private ServerManager(ReactiveWebServerFactory factory) {
    this.handler = this::handleUninitialized;
    this.server = factory.getWebServer(this);
}

public WebServer getWebServer(HttpHandler httpHandler) {
    //I
    HttpServer httpServer = createHttpServer();
    //II
    ReactorHttpHandlerAdapter handlerAdapter = new ReactorHttpHandlerAdapter(
            httpHandler);
    //III
    return new NettyWebServer(httpServer, handlerAdapter, this.lifecycleTimeout);
}

private HttpServer createHttpServer() {
        return HttpServer.builder().options((options) -> {
            options.listenAddress(getListenAddress());
            if (getSsl() != null && getSsl().isEnabled()) {
                SslServerCustomizer sslServerCustomizer = new SslServerCustomizer(
                        getSsl(), getSslStoreProvider());
                sslServerCustomizer.customize(options);
            }
            if (getCompression() != null && getCompression().getEnabled()) {
                CompressionCustomizer compressionCustomizer = new CompressionCustomizer(
                        getCompression());
                compressionCustomizer.customize(options);
            }
            applyCustomizers(options);
        }).build();
    }

private WebServer startReactiveWebServer() {
    ServerManager serverManager = this.serverManager;
    ServerManager.start(serverManager, this::getHttpHandler);
    return ServerManager.getWebServer(serverManager);
}

protected HttpHandler getHttpHandler() {
    // Use bean names so that we don't consider the hierarchy
    String[] beanNames = getBeanFactory().getBeanNamesForType(HttpHandler.class);
    if (beanNames.length == 0) {
        throw new ApplicationContextException(
                "Unable to start ReactiveWebApplicationContext due to missing HttpHandler bean.");
    }
    if (beanNames.length > 1) {
        throw new ApplicationContextException(
                "Unable to start ReactiveWebApplicationContext due to multiple HttpHandler beans : "
                        + StringUtils.arrayToCommaDelimitedString(beanNames));
    }
    return getBeanFactory().getBean(beanNames[0], HttpHandler.class);
}

public static void start(ServerManager manager,
        Supplier<HttpHandler> handlerSupplier) {
    if (manager != null && manager.server != null) {
        manager.handler = handlerSupplier.get();//执行getHttpHandler方法
        manager.server.start();//启动服务
    }
}

public void start() throws WebServerException {
    //IV具体启动服务
    if (this.nettyContext == null) {
        try {
            this.nettyContext = startHttpServer();
        }
        ...
         //开启deamon线程同步等待服务终止
        NettyWebServer.logger.info("Netty started on port(s): " + getPort());
        startDaemonAwaitThread(this.nettyContext);
    }
}

private BlockingNettyContext startHttpServer() {
    if (this.lifecycleTimeout != null) {
        return this.httpServer.start(this.handlerAdapter, this.lifecycleTimeout);
    }
    return this.httpServer.start(this.handlerAdapter);
}

private void startDaemonAwaitThread(BlockingNettyContext nettyContext) {
    //启动线程
    Thread awaitThread = new Thread("server") {

        @Override
        public void run() {
            //同步阻塞服务停止
            nettyContext.getContext().onClose().block();
        }

    };
    //设置线程为demaon，并启动
    awaitThread.setContextClassLoader(getClass().getClassLoader());
    awaitThread.setDaemon(false);
    awaitThread.start();
}

protected void onHandlerStart() {
    applyHandler();
}

protected final void applyHandler() {
    try {
       //1.调用适配器ReactorHttpHandlerAdapter的apply方法
        Mono.fromDirect(handler.apply((INBOUND) this, (OUTBOUND) this))
            .subscribe(this);
    }
    catch (Throwable t) {
        channel.close();
    }
}

public Mono<Void> apply(HttpServerRequest request, HttpServerResponse response) {
    ServerHttpRequest adaptedRequest;
    ServerHttpResponse adaptedResponse;
    try {
        //2.创建与reactor对应的请求、响应对象
        adaptedRequest = new ReactorServerHttpRequest(request, BUFFER_FACTORY);
        adaptedResponse = new ReactorServerHttpResponse(response, BUFFER_FACTORY);
    }
    catch (URISyntaxException ex) {
        ...
        response.status(HttpResponseStatus.BAD_REQUEST);
        return Mono.empty();
    }

    ...
    //3. 这里httpHandler为ServerManager
    return this.httpHandler.handle(adaptedRequest, adaptedResponse)
            .doOnError(ex -> logger.warn("Handling completed with error: " + ex.getMessage()))
            .doOnSuccess(aVoid -> logger.debug("Handling completed with success"));
}

public Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response) {
    //4.这里handler为HttpWebHandlerAdapter
    return this.handler.handle(request, response);
}
接着调用HttpWebHandlerAdapter的handle方法，其代码如下：

public Mono<Void> handle(ServerHttpRequest request, ServerHttpResponse response) {
    //5.创建服务交换对象
    ServerWebExchange exchange = createExchange(request, response);
    //6.这里getDelegate()为DispatcherHandler
    return getDelegate().handle(exchange)
            .onErrorResume(ex -> handleFailure(request, response, ex))
            .then(Mono.defer(response::setComplete));
}

protected ServerWebExchange createExchange(ServerHttpRequest request, ServerHttpResponse response) {
    return new DefaultServerWebExchange(request, response, this.sessionManager,
            getCodecConfigurer(), getLocaleContextResolver(), this.applicationContext);
}

public Mono<Void> handle(ServerWebExchange exchange) {
    ...
    if (this.handlerMappings == null) {
        return Mono.error(HANDLER_NOT_FOUND_EXCEPTION);
    }
    //7.查找对应的controller进行处理
    return Flux.fromIterable(this.handlerMappings)//7.1获取所有处理器映射
    .concatMap(mapping -> mapping.getHandler(exchange))//7.2转换映射，获取处理器
    .next()//7.3获取第一个元素
    .switchIfEmpty(Mono.error(HANDLER_NOT_FOUND_EXCEPTION))//7.4不存在处理器
    .flatMap(handler -> invokeHandler(exchange, handler))//7.5使用处理器进行处理
    .flatMap(result -> handleResult(exchange, result));//7.6处理处理器处理的结果
}