asahi/rikako-note
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
db56eddf842db4171bf9ff17f9ac82e544d5031b
rikako-note/spring/webflux/webclient.md
asahi db56eddf84 doc: 阅读webclient文档
2025-06-03 12:55:01 +08:00

17 KiB
Raw Blame History

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);
connection provider

connection provider会对固定的最大数量连接进行缓存和重用。其中，最大数量针对的是单个connection pool的最大数量，而每个connection pool则是和指定的remote host相关联。

当连接池中的连接都被使用时，后续连接请求将会被阻塞，阻塞时间由pendingAcquireTime决定。

各个超时含义

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

readTimeout

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

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

connectTimeout

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

JDK HttpClient

如果要自定义jdk httpclient配置，欸可以使用如下方式：

HttpClient httpClient = HttpClient.newBuilder()
	.followRedirects(Redirect.NORMAL)
	.connectTimeout(Duration.ofSeconds(20))
	.build();

ClientHttpConnector connector =
		new JdkClientHttpConnector(httpClient, new DefaultDataBufferFactory());

WebClient webClient = WebClient.builder().clientConnector(connector).build();

Jetty

自定义jetty httpclient配置的示例如下所示：

HttpClient httpClient = new HttpClient();
httpClient.setCookieStore(...);

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

HttpComponents

如下示例展示了如何定义Apache HttpComponents：

HttpAsyncClientBuilder clientBuilder = HttpAsyncClients.custom();
clientBuilder.setDefaultRequestConfig(...);
CloseableHttpAsyncClient client = clientBuilder.build();

ClientHttpConnector connector = new HttpComponentsClientHttpConnector(client);

WebClient webClient = WebClient.builder().clientConnector(connector).build();

retrieve()

retrieve()方法用于定义如何对相应进行提取，示例如下：

WebClient client = WebClient.create("https://example.org");

Mono<ResponseEntity<Person>> result = client.get()
		.uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON)
		.retrieve()
		.toEntity(Person.class);

上述示例获取的是ResponseEntity<Person>，如果想要直接获取Persion类型的response body，可以通过如下方式进行获取:

WebClient client = WebClient.create("https://example.org");

Mono<Person> result = client.get()
		.uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON)
		.retrieve()
		.bodyToMono(Person.class);

默认情况下，4xx或5xx的http响应将会导致WebClientResponseException，如果需要自定义error handling逻辑，需要使用onStatus，示例如下：

Mono<Person> result = client.get()
		.uri("/persons/{id}", id).accept(MediaType.APPLICATION_JSON)
		.retrieve()
		.onStatus(HttpStatusCode::is4xxClientError, response -> ...)
		.onStatus(HttpStatusCode::is5xxServerError, response -> ...)
		.bodyToMono(Person.class);

Exchange

exchangeToMono方法和exchangeToFlux方法可以提供更精确的控制，例如在statusCode不同时，使用不同的方法来进行反序列化：

Mono<Person> entityMono = client.get()
		.uri("/persons/1")
		.accept(MediaType.APPLICATION_JSON)
		.exchangeToMono(response -> {
			if (response.statusCode().equals(HttpStatus.OK)) {
				return response.bodyToMono(Person.class);
			}
			else {
				// Turn to error
				return response.createError();
			}
		});

如上所示，当exchangeToMono() and exchangeToFlux()返回的flux或mono完成时，clientResponse对象将会被释放，从而避免内存和连接的泄露。

因而，response不能在更下游进行反序列化，具体的反序列化过程由提供的方法来声明如何进行decode。

Request Body

对于webclient，可以加载任意异步类型对象到request body，例如Mono：

Mono<Person> personMono = ... ;

Mono<Void> result = client.post()
		.uri("/persons/{id}", id)
		.contentType(MediaType.APPLICATION_JSON)
		.body(personMono, Person.class)
		.retrieve()
		.bodyToMono(Void.class);

同样的，也可以加载多个异步对象，例如Flux类型

Flux<Person> personFlux = ... ;

Mono<Void> result = client.post()
		.uri("/persons/{id}", id)
		.contentType(MediaType.APPLICATION_STREAM_JSON)
		.body(personFlux, Person.class)
		.retrieve()
		.bodyToMono(Void.class);

当不想从异步类型的对象加载值，而是想加载已有的值时，可以使用bodyValue方法：

Person person = ... ;

Mono<Void> result = client.post()
		.uri("/persons/{id}", id)
		.contentType(MediaType.APPLICATION_JSON)
		.bodyValue(person)
		.retrieve()
		.bodyToMono(Void.class);

Form Data

如果想要发送form data，可以为body指定MultiValueMap<String, String>类型的值，content会被自动设置为application/x-www-form-urlencoded。使用示例如下所示：

MultiValueMap<String, String> formData = ... ;

Mono<Void> result = client.post()
		.uri("/path", id)
		.bodyValue(formData)
		.retrieve()
		.bodyToMono(Void.class);

同时，可以使用BodyInserters来构建form data，示例如下所示：

Mono<Void> result = client.post()
		.uri("/path", id)
		.body(fromFormData("k1", "v1").with("k2", "v2"))
		.retrieve()
		.bodyToMono(Void.class);

Multipart Data

为了发送multipart data，需要提供MultiValueMap<String, ?>类型的值，其中value类型为代表part内容的object对象或代表part内容和headers的HttpEntity对象。

MutlipartBodyBuilder提供了便捷的api，使用示例如下所示：

MultipartBodyBuilder builder = new MultipartBodyBuilder();
builder.part("fieldPart", "fieldValue");
builder.part("filePart1", new FileSystemResource("...logo.png"));
builder.part("jsonPart", new Person("Jason"));
builder.part("myPart", part); // Part from a server request

MultiValueMap<String, HttpEntity<?>> parts = builder.build();

通常情况下，无需为每个part指定Content-Type，在HttpMessageWriter执行序列化操作时，会自动的设置ContentType类型，对于Resources类型，会根据文件拓展名来决定ContentType。

当MutliValueMap类型的值构建完成后，使用示例如下：

MultipartBodyBuilder builder = ...;

Mono<Void> result = client.post()
		.uri("/path", id)
		.body(builder.build())
		.retrieve()
		.bodyToMono(Void.class);

除了使用MultipartBodyBuilder之外，还可以使用BodyInserters来构建multipart body，示例如下：

Mono<Void> result = client.post()
		.uri("/path", id)
		.body(fromMultipartData("fieldPart", "value").with("filePart", resource))
		.retrieve()
		.bodyToMono(Void.class);

PartEvent

如果需要提供多个multipart data，可以使用PartEvent类型：

  • form fields可以通过FormPartEvent::create来创建
  • file uploads可以通过FormPartEvent::create来创建

可以通过Flux.concat方法来对其进行拼接，示例如下所示：

Resource resource = ...
Mono<String> result = webClient
	.post()
	.uri("https://example.com")
	.body(Flux.concat(
			FormPartEvent.create("field", "field value"),
			FilePartEvent.create("file", resource)
	), PartEvent.class)
	.retrieve()
	.bodyToMono(String.class);

Filters

可以通过WebClient.Builder注册client filter，其可以针对请求进行拦截和修改，示例如下所示：

WebClient client = WebClient.builder()
		.filter((request, next) -> {

			ClientRequest filtered = ClientRequest.from(request)
					.header("foo", "bar")
					.build();

			return next.exchange(filtered);
		})
		.build();

其可被用作认证，示例如下：

WebClient client = WebClient.builder()
		.filter(basicAuthentication("user", "password"))
		.build();

通过修改WebClient对象，filter可被添加和删除，修改后的新webClient对象并不会影响之前的对象，示例如下：

WebClient client = webClient.mutate()
		.filters(filterList -> {
			filterList.add(0, basicAuthentication("user", "password"));
		})
		.build();

filter处理response

在使用filter时，需要确保response content总是被消费：

  • 当filter在处理response时，需要确保filter总是：
    • 消费response content
    • 或将response传递到下游

示例代码如下所示：

public ExchangeFilterFunction renewTokenFilter() {
	return (request, next) -> next.exchange(request).flatMap(response -> {
		if (response.statusCode().value() == HttpStatus.UNAUTHORIZED.value()) {
			return response.releaseBody()
					.then(renewToken())
					.flatMap(token -> {
						ClientRequest newRequest = ClientRequest.from(request).build();
						return next.exchange(newRequest);
					});
		} else {
			return Mono.just(response);
		}
	});
}

如下示例创建了自定义的filter，用于计算POST和PUT的multipart/form-data请求的Content-Length

public class MultipartExchangeFilterFunction implements ExchangeFilterFunction {

	@Override
	public Mono<ClientResponse> filter(ClientRequest request, ExchangeFunction next) {
		if (MediaType.MULTIPART_FORM_DATA.includes(request.headers().getContentType())
				&& (request.method() == HttpMethod.PUT || request.method() == HttpMethod.POST)) {
			return next.exchange(ClientRequest.from(request).body((outputMessage, context) ->
				request.body().insert(new BufferingDecorator(outputMessage), context)).build()
			);
		} else {
			return next.exchange(request);
		}
	}

	private static final class BufferingDecorator extends ClientHttpRequestDecorator {

		private BufferingDecorator(ClientHttpRequest delegate) {
			super(delegate);
		}

		@Override
		public Mono<Void> writeWith(Publisher<? extends DataBuffer> body) {
			return DataBufferUtils.join(body).flatMap(buffer -> {
				getHeaders().setContentLength(buffer.readableByteCount());
				return super.writeWith(Mono.just(buffer));
			});
		}
	}
}

Attributes

可以向request中添加attributes，并且可以通过attributes来影响filter中的行为

WebClient client = WebClient.builder()
		.filter((request, next) -> {
			Optional<Object> usr = request.attribute("myAttribute");
			// ...
		})
		.build();

client.get().uri("https://example.org/")
		.attribute("myAttribute", "...")
		.retrieve()
		.bodyToMono(Void.class);

	}

除此之外，可以通过org.springframework.web.reactive.function.client.DefaultWebClientBuilder#defaultRequest方法为webClient注册一个callback回调，该回调可为所有请求插入attribute。