springboot(六)——springboot與webflux結合初探

這幾天看了看spring-cloud-gateway的請求處理流程，因爲之前一直用的springboot1.x和spring4，一開始對spring-cloud-gateway的處理流程有點懵逼，找不到入口，後來跟了代碼，在網上找了點資料，發現spring-cloud-gateway的入口在ReactorHttpHandlerAdapter的apply方法

public class ReactorHttpHandlerAdapter implements BiFunction<HttpServerRequest, HttpServerResponse, Mono<Void>> {

	private static final Log logger = HttpLogging.forLogName(ReactorHttpHandlerAdapter.class);


	private final HttpHandler httpHandler;
	public ReactorHttpHandlerAdapter(HttpHandler httpHandler) {
		Assert.notNull(httpHandler, "HttpHandler must not be null");
		this.httpHandler = httpHandler;
	}
	@Override
	public Mono<Void> apply(HttpServerRequest reactorRequest, HttpServerResponse reactorResponse) {
		NettyDataBufferFactory bufferFactory = new NettyDataBufferFactory(reactorResponse.alloc());
		try {
			ReactorServerHttpRequest request = new ReactorServerHttpRequest(reactorRequest, bufferFactory);
			ServerHttpResponse response = new ReactorServerHttpResponse(reactorResponse, bufferFactory);

			if (request.getMethod() == HttpMethod.HEAD) {
				response = new HttpHeadResponseDecorator(response);
			}

			return this.httpHandler.handle(request, response)
					.doOnError(ex -> logger.trace(request.getLogPrefix() + "Failed to complete: " + ex.getMessage()))
					.doOnSuccess(aVoid -> logger.trace(request.getLogPrefix() + "Handling completed"));
		}
		catch (URISyntaxException ex) {
			if (logger.isDebugEnabled()) {
				logger.debug("Failed to get request URI: " + ex.getMessage());
			}
			reactorResponse.status(HttpResponseStatus.BAD_REQUEST);
			return Mono.empty();
		}
	}

}

該方法的作用就是把接收到的HttpServerRequest或者最終需要返回的HttpServerResponse，包裝轉換爲ReactorServerHttpRequest和ReactorServerHttpResponse。

spring-webflux

當然，這篇文章的主要內容不是談論spring-cloud-gateway了，因爲之前一直用的spring4，所以對spring5當中的反應式編程範式和webflux不太瞭解，所以先寫個demo瞭解一下

第一步：引入相關pom,測試的相關pom根據自己的需要引入

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.1.4.RELEASE</version>
    <relativePath/> <!-- lookup parent from repository -->
</parent>


<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-webflux</artifactId>
    </dependency>

    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>io.projectreactor</groupId>
        <artifactId>reactor-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

第二步：創建一個HandlerFunction

public class TestFunction implements HandlerFunction<ServerResponse> {

    @Override
    public Mono<ServerResponse> handle(ServerRequest serverRequest) {
        return ServerResponse.ok().body(
                Mono.just(parse(serverRequest, "args1") + parse(serverRequest, "args2"))
                , Integer.class);
    }

    private int parse(final ServerRequest request, final String param) {
        return Integer.parseInt(request.queryParam(param).orElse("0"));
    }
}

第三步：注入一個RouterFunction

@Configuration
public class TestRouteFunction {

    @Bean
    public RouterFunction<ServerResponse> routerFunction() {
        return RouterFunctions.route(RequestPredicates.GET("/add"), new TestFunction());
    }
}

第四步：在webflux中，也可以使用之前的java註解的編程方式，我們也創建一個controller

@RestController
@RequestMapping("/api/test")
public class HelloController {

    @RequestMapping("/hello")
    public Mono<String> hello() {
        return Mono.just("hello world");
    }
}

第五步：創建啓動類

@SpringBootApplication
public class Spring5DemoApplication {

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

第六步：啓動項目，訪問如下兩個接口都可以

http://localhost:8080/api/test/hello
http://localhost:8080/add?args1=2&args2=3

和spring-boot結合

通過上面的例子，我們看到基本的兩個類：HandlerFunction和RouterFunction，同時webflux有如下特性：

異步非阻塞
響應式(reactive)函數編程，純lambda表達式

不僅僅是在Servlet容器中tomcat/jetty中運行，同時支持NIO的Netty和Undertow中，實際項目中，我們往往與spring-boot項目結合，我們跟進代碼可以看看spring-boot是在什麼時候創建的server
一、SpringApplication

public ConfigurableApplicationContext run(String... args) {
	StopWatch stopWatch = new StopWatch();
	stopWatch.start();
	ConfigurableApplicationContext context = null;
	Collection<SpringBootExceptionReporter> exceptionReporters = new ArrayList<>();
	configureHeadlessProperty();
	SpringApplicationRunListeners listeners = getRunListeners(args);
	listeners.starting();
	try {
		ApplicationArguments applicationArguments = new DefaultApplicationArguments(
				args);
		ConfigurableEnvironment environment = prepareEnvironment(listeners,
				applicationArguments);
		configureIgnoreBeanInfo(environment);
		Banner printedBanner = printBanner(environment);
		context = createApplicationContext();
		exceptionReporters = getSpringFactoriesInstances(
				SpringBootExceptionReporter.class,
				new Class[] { ConfigurableApplicationContext.class }, context);
		prepareContext(context, environment, listeners, applicationArguments,
				printedBanner);
		refreshContext(context);
		afterRefresh(context, applicationArguments);
		stopWatch.stop();
		if (this.logStartupInfo) {
			new StartupInfoLogger(this.mainApplicationClass)
					.logStarted(getApplicationLog(), stopWatch);
		}
		listeners.started(context);
		callRunners(context, applicationArguments);
	}
	catch (Throwable ex) {
		handleRunFailure(context, ex, exceptionReporters, listeners);
		throw new IllegalStateException(ex);
	}

	try {
		listeners.running(context);
	}
	catch (Throwable ex) {
		handleRunFailure(context, ex, exceptionReporters, null);
		throw new IllegalStateException(ex);
	}
	return context;
}

我們只分析入口，其它代碼暫時不管，找到refreshContext(context);這一行進去

二、ReactiveWebServerApplicationContext的refresh()

@Override
public final void refresh() throws BeansException, IllegalStateException {
	try {
		super.refresh();
	}
	catch (RuntimeException ex) {
		stopAndReleaseReactiveWebServer();
		throw ex;
	}
}

三、ReactiveWebServerApplicationContext的onRefresh()

@Override
protected void onRefresh() {
	super.onRefresh();
	try {
		createWebServer();
	}
	catch (Throwable ex) {
		throw new ApplicationContextException("Unable to start reactive web server",
				ex);
	}
}

四、看到這裏我們就找到入口方法了：createWebServer(),跟進去，找到NettyReactiveWebServerFactory中創建webserver

@Override
public WebServer getWebServer(HttpHandler httpHandler) {
	HttpServer httpServer = createHttpServer();
	ReactorHttpHandlerAdapter handlerAdapter = new ReactorHttpHandlerAdapter(
			httpHandler);
	return new NettyWebServer(httpServer, handlerAdapter, this.lifecycleTimeout);
}

看到ReactorHttpHandlerAdapter這個類想必特別親切，在開篇說過是spring-cloud-gateway的入口，createHttpServer方法的細節暫時沒有去學習了，後續有時間去深入瞭解下