Spring—WebClient使用

阅读完需：约 9 分钟

WebClient是Spring5引入的，基于响应式编程实现的HTTP调用客户端。Spring官方推荐使用WebClient替代RestTemplate完成HTTP调用。因为WebClient是基于Reactor实现的，所以既可以支持阻塞调用也可以支持非阻塞调用，在高并发的场景下资源利用率更高。这也是官方推荐使用的重要原因之一。

WebClient 内部委托给HTTP客户端库。默认情况下，WebClient 使用 Reactor Netty

如果在工程中想要是用WebClient，在Pom文件中加入如下依赖

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

SpringBoot版本：3.0.2

初始化WebClient

方式一：通过静态工厂方法创建响应式WebClient实例

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

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

//响应式客户端
WebClient client = null;
String baseUrl = "http://127.0.0.1:1112/greeting";
client = WebClient.create(baseUrl);
/**
 * 是通过 WebClient 组件构建请求
 */
String restUrl = baseUrl + "api/demo/hello/v1";
Mono<String> stringMono = client
        // 请求方法
        .method(HttpMethod.GET)
        // 请求url 和 参数
        //.uri(restUrl, params)
        .uri(restUrl)
        // 媒体的类型
        .accept(MediaType.APPLICATION_JSON).retrieve().bodyToMono(String.class);

@Test
public void testGet() throws IOException, InterruptedException {

	Mono<String> resp = WebClient.create()
			.method(HttpMethod.GET)
			.uri("http://127.0.0.1:1112/greeting")
			.cookie("token", "jwt_token")
			.header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
			.retrieve().bodyToMono(String.class);

	// 订阅结果
	resp.subscribe(responseData ->
	{
		System.out.println("qqqqq"+responseData.toString());
	}, e ->
	{
		System.out.println("error:" + e.getMessage());
	});
	//主线程等待， 一切都是为了查看到异步结果
	Thread.sleep(2000);
}

方式二：使用builder（构造者）创建响应式WebClient实例

WebClient.builder().build();

WebClient client4 = WebClient.builder()
        .baseUrl("http://localhost:8080")
        .defaultCookie("cookieKey", "cookieValue")
        .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
        .defaultUriVariables(Collections.singletonMap("url", "http://localhost:8080"))
        .build();

使用build来构建主要是为了自定义参数

还可以使用WebClient.builder()其他选项：

• uriBuilderFactory：自定义UriBuilderFactory用作基本URL(BaseUrl)。
• defaultHeader：每个请求的标题。
• defaultCookie：针对每个请求的Cookie。
• defaultRequest：Consumer自定义每个请求。
• filter：针对每个请求的客户端过滤器。
• exchangeStrategies：HTTP消息读取器/写入器定制。
• clientConnector：HTTP客户端库设置。

方式三：WebClient实例克隆

一旦建立，WebClient实例是不可变的。但是，您可以克隆它并构建修改后的副本，而不会影响原始实例，如以下示例所示：

WebClient client1 = WebClient.builder().build();
WebClient client2 = client1.mutate().build();

主要是mutate()函数，返回一个构建器以创建一个新的WebClient，其设置是从当前WebClient复制的。

总结一下上面三种方式都是和DefaultWebClientBuilder类有关，本质上都是相同的。

第一种

WebClient.create()

	/**
	 * Create a new {@code WebClient} with Reactor Netty by default.
	 * @see #create(String)
	 * @see #builder()
	 */
	static WebClient create() {
		return new DefaultWebClientBuilder().build();
	}

第二种

WebClient.builder().build();

	/**
	 * Obtain a {@code WebClient} builder.
	 */
	static WebClient.Builder builder() {
		return new DefaultWebClientBuilder();
	}

第三种

client1.mutate().build();

	@Override
	public Builder mutate() {
		return new DefaultWebClientBuilder(this.builder);
	}

可以通过builder()方式来构建自定义参数，修改默认的超时时间

//通过HttpClient设置超时时间
HttpClient httpClient2 = HttpClient.create()
		//设置连接超时时间
		.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
		//设置响应超时时间
		.responseTimeout(Duration.ofMillis(5000))
		//分别设置读写超时时间
		.doOnConnected(conn -> conn.addHandlerLast(new ReadTimeoutHandler(5000, TimeUnit.MILLISECONDS))     .addHandlerLast(new WriteTimeoutHandler(5000,TimeUnit.MILLISECONDS)));

WebClient client5 = WebClient.builder()
		.clientConnector(new ReactorClientHttpConnector(httpClient2))
		.build();

请求提交

GET请求

Mono<String> resp = WebClient.create()
		.method(HttpMethod.GET)
		.uri("http://127.0.0.1:1112/greeting")
		.cookie("token", "jwt_token")
		.header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
		//获取结果
		.retrieve()
		//将结果转化为指定类型
		.bodyToMono(String.class);

// 通过非阻塞的方式获取响应结果 订阅结果 
resp.subscribe(responseData ->
{
	System.out.println("qqqqq"+responseData.toString());
}, e ->
{
	System.out.println("error:" + e.getMessage());
});
//主线程等待， 一切都是为了查看到异步结果
Thread.sleep(2000);

//通过阻塞的方式获取响应结果
String block = resp.block();

POST请求

提交Json Body

Mono<Person> personMono = ... ;

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

例子

@Test
public void testJSONParam() throws InterruptedException {
	LoginInfoDTO dto=new LoginInfoDTO("lisi","123456");
	Mono<LoginInfoDTO> personMono =Mono.just(dto);

	Mono<String> resp = WebClient.create().post()
			.uri("http://127.0.0.1:1112/greeting")
			.contentType(MediaType.APPLICATION_JSON)
			//.contentType(MediaType.APPLICATION_FORM_URLENCODED)
			.body(personMono,LoginInfoDTO.class)
			.retrieve()
			.bodyToMono(String.class);

	// 订阅结果
	resp.subscribe(responseData ->
	{
		System.out.println(responseData.toString());
	}, e ->
	{
		System.out.println("error:" + e.getMessage());
	});
	//主线程等待， 一切都是为了查看到异步结果
	Thread.sleep(2000);
}

提交表单

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

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

或者

import static org.springframework.web.reactive.function.BodyInserters.*;

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

例子

MultiValueMap<String, String> formData = new LinkedMultiValueMap<>();
		formData.add("name1","value1");
		formData.add("name2","value2");
		Mono<String> resp2 = WebClient.create().post()
				.uri("http://localhost:8080/submit")
				.contentType(MediaType.APPLICATION_FORM_URLENCODED)
				.body(BodyInserters.fromFormData(formData))
				.retrieve()
				.bodyToMono(String.class);
		logger.info("result:{}",resp.block());

上传文件

@Test
public void testUploadFile()
{
	HttpHeaders headers = new HttpHeaders();
	headers.setContentType(MediaType.IMAGE_PNG);
	HttpEntity<ClassPathResource> entity =
			new HttpEntity<>(new ClassPathResource("logback-spring.xml"), headers);
	MultiValueMap<String, Object> parts = new LinkedMultiValueMap<>();
	parts.add("file", entity);
	Mono<String> resp = WebClient.create().post()
			.uri("http://127.0.0.1:1112/greeting")
			.contentType(MediaType.MULTIPART_FORM_DATA)
			.body(BodyInserters.fromMultipartData(parts))
			.retrieve().bodyToMono(String.class);
	System.out.println("result:"+resp.block());
}

错误处理

• 可以使用onStatus根据status code进行异常适配
• 可以使用doOnError异常适配
• 可以使用onErrorReturn返回默认值

@Test
public void testFormParam4xx()
{
	WebClient webClient = WebClient.builder()
			.baseUrl("https://api.github.com")
			.defaultHeader(HttpHeaders.CONTENT_TYPE, "application/vnd.github.v3+json")
			.defaultHeader(HttpHeaders.USER_AGENT, "Spring 5 WebClient")
			.build();
	WebClient.ResponseSpec responseSpec = webClient.method(HttpMethod.GET)
			.uri("/user/repos?sort={sortField}&direction={sortDirection}", "updated", "desc")
			.retrieve();
	Mono<String> mono = responseSpec
			.onStatus(e -> e.is4xxClientError(), resp ->
			{
				log.info("error:{},msg:{}", resp.statusCode().value(), resp.statusCode().getReasonPhrase());
				return Mono.error(new RuntimeException(resp.statusCode().value() + " : " + resp.statusCode().getReasonPhrase()));
			})
			.bodyToMono(String.class)
			.doOnError(WebClientResponseException.class, err ->
			{
				log.info("ERROR status:{},msg:{}", err.getRawStatusCode(), err.getResponseBodyAsString());
				throw new RuntimeException(err.getMessage());
			})
			.onErrorReturn("fallback");
	String result = mono.block();
	System.out.print(result);
}

响应解码

有两种对响应的处理方法：

• retrieveretrieve方法是直接获取响应body。
• exchange但是，如果需要响应的头信息、Cookie等，可以使用exchange方法，exchange方法可以访问整个ClientResponse。

异步转同步：

由于响应的得到是异步的，所以都可以调用 block 方法来阻塞当前程序，等待获得响应的结果。

retrieve

该retrieve()方法是获取响应主体并对其进行解码的最简单方法。

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

默认情况下，4XX或5xx状态代码的应答导致 WebClientResponseException或它的HTTP状态的具体子类之一，比如 WebClientResponseException.BadRequest，WebClientResponseException.NotFound和其他人。还可以使用该onStatus方法来自定义所产生的异常

exchange()

该exchange()方法比该方法提供更多的控制retrieve。以下示例等效于retrieve()但也提供对的访问ClientResponse：

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

请注意与不同retrieve(），对于exchange()，没有4xx和5xx响应的自动错误信号。您必须检查状态码并决定如何进行。

与相比retrieve()，当使用时exchange()，应用程序有责任使用任何响应内容，而不管情况如何（成功，错误，意外数据等）,否则会导致内存泄漏.

例子

/**
 * 测试用例： Exchange
 */
@Test
public void testExchange()
{
	String baseUrl = "http://localhost:8081";
	WebClient webClient = WebClient.create(baseUrl);

	MultiValueMap<String, String> map = new LinkedMultiValueMap<>();
	map.add("username", "u123");
	map.add("password", "p123");

	Mono<ClientResponse> loginMono = webClient.post().uri("login").syncBody(map).exchange();
	ClientResponse response = loginMono.block();
	if (response.statusCode() == HttpStatus.OK) {
		Mono<RestOut> resultMono = response.bodyToMono(RestOut.class);
		resultMono.subscribe(result -> {
			if (result.isSuccess()) {
				ResponseCookie sidCookie = response.cookies().getFirst("sid");
				Mono<LoginInfoDTO> dtoMono = webClient.get().uri("users").cookie(sidCookie.getName(), sidCookie.getValue()).retrieve().bodyToMono(LoginInfoDTO.class);
				dtoMono.subscribe(System.out::println);
			}
		});
	}
}

response body 转换响应流

将response body 转换为对象/集合

• bodyToMono如果返回结果是一个Object，WebClient将接收到响应后把JSON字符串转换为对应的对象，并通过Mono流弹出。
• bodyToFlux如果响应的结果是一个集合，则不能继续使用bodyToMono()，应该改用bodyToFlux()，然后依次处理每一个元素，并通过Flux流弹出。

请求和响应过滤

WebClient也提供了Filter，对应于org.springframework.web.reactive.function.client.ExchangeFilterFunction接口，其接口方法定义如下。

Mono<ClientResponse> filter(ClientRequest request, ExchangeFunction next)

在进行拦截时可以拦截request，也可以拦截response。

WebClient webClient = WebClient.builder()
		.baseUrl("http://localhost:8081")
		.defaultHeader(HttpHeaders.CONTENT_TYPE, "application/vnd.github.v3+json")
		.filter(ExchangeFilterFunctions
				.basicAuthentication(username, token))
		.build();

例子

@Test
void filter() {
	Map<String, Object> uriVariables = new HashMap<>();
	uriVariables.put("p1", "var1");
	uriVariables.put("p2", 1);
	WebClient webClient = WebClient.builder().baseUrl("http://www.xxxxx.com")
			.filter(logResposneStatus())
			.defaultHeader(HttpHeaders.CONTENT_TYPE, "application/vnd.github.v3+json")
			.build();
	Mono<String> resp1 = webClient
			.method(HttpMethod.GET)
			.uri("/")
			.cookie("token","xxxx")
			.header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
			.retrieve().bodyToMono(String.class);
	String re=  resp1.block();
	System.out.print("result:" +re);

}

private ExchangeFilterFunction logResposneStatus() {
	return ExchangeFilterFunction.ofResponseProcessor(clientResponse -> {
		log.info("Response Status {}", clientResponse.statusCode());
		return Mono.just(clientResponse);
	});
}