rikako-note/spring/webflux/webclient.md

WebClient

WebClient基于Reactor提供了functional, fluent API。

WebClient是非阻塞的，其依赖的codecs和server端使用的codecs相同。

Configuration

创建WebClient最简单的方式是通过静态工厂方法：

• WebClient.create()
• WebClient.create(String baseUrl)

除此之外，也可以通过WebClient.builder()来指定更多选项：

• uriBuilderFactory： 自定义uriBuilderFactory，用于创建UriBuilder，UriBuilder包含共享的配置，例如base URI等
• defaultUriVariables: 在拓展uri templates时，使用到的默认值
• defaultHeader：对每个请求都包含的headers
• defaultCookie：每个请求都包含的Cookie
• defaultRequest： 对每个请求进行自定义的Consumer
• filter：对于每个请求的client filter
• exchangeStrategies：自定义http message的reader/writer
• clientConnector：http client library设置
• observationRegistry： the registry to use for enabling Observability support
• observationConvention: an optional, custom convention to extract metadata for recorded observations.

创建WebClient的示例如下：

WebClient client = WebClient.builder()
		.codecs(configurer -> ... )
		.build();

一旦被创建后，WebClient是不可变的，但是，可以对其进行克隆并对副本进行修改，示例如下：

WebClient client1 = WebClient.builder()
		.filter(filterA).filter(filterB).build();

WebClient client2 = client1.mutate()
		.filter(filterC).filter(filterD).build();

// client1 has filterA, filterB

// client2 has filterA, filterB, filterC, filterD

MaxInMemorySize

Codecs对于缓存在内存中的数据大小存在限制，避免导致应用的内存问题。默认情况下，该值被设置为256KB，如果该大小不够，那么将会见到如下报错：

org.springframework.core.io.buffer.DataBufferLimitException: Exceeded limit on max bytes to buffer

如果想要修改该默认的codecs限制，可以使用如下配置：

WebClient webClient = WebClient.builder()
		.codecs(configurer -> configurer.defaultCodecs().maxInMemorySize(2 * 1024 * 1024))
		.build();

reactor netty

为了自定义reactor netty配置，可以提供一个预先定义的HttpClient

HttpClient httpClient = HttpClient.create().secure(sslSpec -> ...);

WebClient webClient = WebClient.builder()
		.clientConnector(new ReactorClientHttpConnector(httpClient))
		.build();

Resources

默认情况下，HttpClient将会参与使用reactor.netty.http.HttpResources中保存的全局reactor netty resources。

HttpResources中包含event loop threads和connection pool，HttpClient会使用这些共享的资源。对于基于event loop的并发，更倾向使用固定、共享的资源，直到进程退出时这些全局资源才会被释放。

如果server的生命周期和进程相同，那么对全局资源无需手动释放。但若server在进程的生命周期中可能会启动和停止，那么可以声明一个ReactorResourceFactory类型的bean，并设置globalResources=true（默认），从而确保reactor netty的global resource在ApplicationContext关闭时被释放，示例如下所示:

@Bean
public ReactorResourceFactory reactorResourceFactory() {
	return new ReactorResourceFactory();
}

也可以不使用global reactor netty resources，但是，在该模式下，你需要确保所有reactor netty client and server使用共享的资源，示例如下所示：

@Bean
public ReactorResourceFactory resourceFactory() {
	ReactorResourceFactory factory = new ReactorResourceFactory();
	factory.setUseGlobalResources(false); 
	return factory;
}

@Bean
public WebClient webClient() {

	Function<HttpClient, HttpClient> mapper = client -> {
		// Further customizations...
	};

	ClientHttpConnector connector =
			new ReactorClientHttpConnector(resourceFactory(), mapper); 

	return WebClient.builder().clientConnector(connector).build(); 
}

timeout

如果要设置connection timeout，示例如下：

import io.netty.channel.ChannelOption;

HttpClient httpClient = HttpClient.create()
		.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000);

WebClient webClient = WebClient.builder()
		.clientConnector(new ReactorClientHttpConnector(httpClient))
		.build();

如果要配置read/write timeout，可以按照如下示例进行配置：

import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.handler.timeout.WriteTimeoutHandler;

HttpClient httpClient = HttpClient.create()
		.doOnConnected(conn -> conn
				.addHandlerLast(new ReadTimeoutHandler(10))
				.addHandlerLast(new WriteTimeoutHandler(10)));

// Create WebClient...

readTimeoutHandler

在指定的时间内，如果channel没有执行读取操作，将会抛出ReadTimeoutException，并对channel执行关闭操作

没有数据被读取的监控，是通过IdleStateHandler来实现的。

ReadTimeoutHandler继承了IdleStateHandler，当指定的readIdleTime过后，如果channel仍然没有指定任何的读操作，那么将会触发一个IdleStateEvent事件，并且事件状态为IdleState.READER_IDLE

writeTimeoutHandler

而writeTimeoutHandler作用则如下：

• 其继承了ChannelOutboundHandlerAdapter，在通过write写入数据时，其会通过ctx.executor()指定一个WriteTimeoutTask，该task在指定timeout后被执行
• 如果在writeTimeoutTask被执行时，写操作仍然没有处理完成，那么其将会触发一个WriteTimeoutException，并且channel也会被关闭。

为所有请求配置response timeout:

HttpClient httpClient = HttpClient.create()
		.responseTimeout(Duration.ofSeconds(2));

// Create WebClient...

为指定的请求配置response timeout:

WebClient.create().get()
		.uri("https://example.org/path")
		.httpRequest(httpRequest -> {
			HttpClientRequest reactorRequest = httpRequest.getNativeRequest();
			reactorRequest.responseTimeout(Duration.ofSeconds(2));
		})
		.retrieve()
		.bodyToMono(String.class);

各个超时含义

再restTemplate等阻塞api中，各个超时的含义如下所示：

readTimeout

会为serverSocket（和server进行通信的socket）设置SocketOptions.SO_SOCKET，单位为ms。

当SO_SOCKET被设置为正数值时，对socket关联的inputstream阻塞式调用，将最多阻塞该指定时长。如果调用超时，那么将会抛出java. net. SocketTimeoutException，但是socket仍然有效。

connectTimeout

以ms为单位指定一个超时时长，当与服务端建立连接时，如果超过该时长后连接仍未建立，那么将会抛出java.net.SocketTimeoutException异常。