Spring5.0 — WebClient（响应式web客户端）

一、介绍

1.1、RestTemplate 

同步阻塞代码，http 请求返回响应才继续执行。

1.2、WebClient 

1.基于 Reactor 和 Netty。
2.响应式 web 客户端。异步执行不阻塞代码，少量的线程数处理高并发的 Http 请求。
3.集成 Spring WebFlux 框架，可与其他 Spring 组件无缝协作。
4.可通过自定义 ExchangeFilterFunction 对请求和响应进行拦截和处理。

二、使用 

2.1、引入依赖 

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

2.2、基础属性介绍

2.2.1、基础配置

HTTP 底库 // 选择 HTTP 底库; 默认底层用Netty，切换Jetty。 WebClient          .builder()          .clientConnector(new JettyClientHttpConnector()) 		 .build();   全局的请求配置 // 设置基础的全局的web请求配置，如cookie、header、baseUrl。 WebClient 		.builder() 		.defaultCookie("kl","kl") 		.defaultUriVariables(ImmutableMap.of("name","kl")) 		.defaultHeader("header","kl") 		.defaultHeaders(httpHeaders -> { 			httpHeaders.add("header1","kl"); 			httpHeaders.add("header2","kl"); 		}) 		.defaultCookies(cookie ->{ 			cookie.add("cookie1","kl"); 			cookie.add("cookie2","kl"); 		}) 		.baseUrl("http://www.kailing.pub") 		.build();   Filter // Filter 过滤器，统一修改拦截请求。 WebClient 	.builder() 	.baseUrl("http://www.kailing.pub") 	.filter((request, next) -> { 		ClientRequest filtered = ClientRequest.from(request) 				.header("foo", "bar") 				.build(); 		return next.exchange(filtered); 	}) 	.filters(filters ->{ 		filters.add(ExchangeFilterFunctions.basicAuthentication("username","password")); 		filters.add(ExchangeFilterFunctions.limitResponseSize(800)); 	}) 	.build().get() 	.uri("/article/index/arcid/{id}.html", 254) 	.retrieve() 	.bodyToMono(String.class) 	.subscribe(System.err::println);   Netty 库配置  // 配置动态连接池 ConnectionProvider provider = ConnectionProvider.elastic("elastic pool"); 配置固定大小连接池，如最大连接数、连接获取超时、空闲连接死亡时间等 ConnectionProvider provider = ConnectionProvider.fixed("fixed", 45, 4000, Duration.ofSeconds(6)); HttpClient httpClient = HttpClient.create(provider) 		.secure(sslContextSpec -> { 			SslContextBuilder sslContextBuilder = SslContextBuilder.forClient().trustManager(new File("E://server.truststore")); 			sslContextSpec.sslContext(sslContextBuilder); 		}).tcpConfiguration(tcpClient -> { 			// 指定Netty的 select 和 work 线程数量 			LoopResources loop = LoopResources.create("kl-event-loop", 1, 4, true); 			return tcpClient.doOnConnected(connection -> { 				// 读写超时设置 				connection 				.addHandlerLast(new ReadTimeoutHandler(10, TimeUnit.SECONDS)) 				.addHandlerLast(new WriteTimeoutHandler(10)); 			}) 				// 连接超时设置 			   .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000) 			   .option(ChannelOption.TCP_NODELAY, true) 			   .runOn(loop); 		}); WebClient.builder() 		.clientConnector(new ReactorClientHttpConnector(httpClient)) 		.build();

2.2.2、WebClient.builder() 的选项

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

2.2.3、请求类型、与返回结果

.block()    阻塞当前程序等待结果 .retrieve()    直接获取响应body .exchange()    可访问整个ClientResponse 

2.3、get 请求

// 简单传参。默认异步，通过 Mono 的 subscribe 订阅响应值。block() 阻塞当前线程获取返回值。 WebClient client = WebClient.create("http://www.kailing.pub"); Mono<String> result = client.get() 		.uri("/article/arcid/{id}", 256) 		.acceptCharset(StandardCharsets.UTF_8) 		.accept(MediaType.TEXT_HTML) 		.retrieve()     // 同步 		.bodyToMono(String.class); result.subscribe(System.err::println);   // 复杂传参 — MultiValueMap。默认异步，通过 Mono 的 subscribe 订阅响应值。block() 阻塞当前线程获取返回值。 MultiValueMap<String, String> params = new LinkedMultiValueMap<>(); params.add("name", "kl"); params.add("age", "19"); // 定义 url 参数  Map<String, Object> uriVariables = new HashMap<>();  uriVariables.put("id", 200); String uri = UriComponentsBuilder.fromUriString("/article/arcid/{id}") 		.queryParams(params) 		.uriVariables(uriVariables) 		.toUriString(); Mono<String> result = client.get() 		.uri(uri) 		.acceptCharset(StandardCharsets.UTF_8) 		.accept(MediaType.TEXT_HTML) 		.retrieve() 		.bodyToMono(String.class); result.subscribe(System.err::println);   // 复杂传参 — UriBuilder。默认异步，通过 Mono 的 subscribe 订阅响应值。block() 阻塞当前线程获取返回值。 Mono<String> resp = WebClient.create() 			.get() 			.uri(uriBuilder -> uriBuilder 					.scheme("http") 					.host("www.baidu.com") 					.path("/s") 					.queryParam("wd", "北京天气") 					.queryParam("other", "test") 					.build()) 			.retrieve() 			.bodyToMono(String.class);

2.4、post 请求

// 表单 默认异步，通过 Mono 的 subscribe 订阅响应值。block() 阻塞当前线程获取返回值。 MultiValueMap<String, String> formData = new LinkedMultiValueMap<>(); 	formData.add("name1","value1"); 	formData.add("name2","value2"); 	Mono<String> resp = WebClient.create().post() 			.uri("http://www.w3school.com.cn/test/demo_form.asp") 			.contentType(MediaType.APPLICATION_FORM_URLENCODED) 			.body(BodyInserters.fromFormData(formData)) 			.retrieve().bodyToMono(String.class); 	LOGGER.info("result:{}",resp.block());  // FormInserter  表单：参数、文件。默认异步，通过 Mono 的 subscribe 订阅响应值。block() 阻塞当前线程获取返回值。 WebClient client = WebClient.create("http://www.kailing.pub"); FormInserter formInserter = fromMultipartData("name","kl") 		.with("age",19) 		.with("map",ImmutableMap.of("xx","xx")) 		.with("file",new File("C://xxx.doc")); Mono<String> result = client.post() 		.uri("/article/index/arcid/{id}.html", 256) 		.contentType(MediaType.APPLICATION_JSON) 		.body(formInserter) 		//.bodyValue(ImmutableMap.of("name","kl")) 		.retrieve() 		.bodyToMono(String.class); result.subscribe(System.err::println);   // json — 实体类。默认异步，通过 Mono 的 subscribe 订阅响应值。block() 阻塞当前线程获取返回值。 User user = new User(); user.setName("aaa"); user.setTitle("AAAAAA"); Mono<String> resp = WebClient.create()     .post() 	.uri("http://localhost:8080/demo/json") 	.contentType(MediaType.APPLICATION_JSON_UTF8) 	.body(Mono.just(user),User.class) 	.retrieve().bodyToMono(String.class); System.out.println("---resp.block(): "+resp.block());   // json — raw。默认异步，通过 Mono 的 subscribe 订阅响应值。block() 阻塞当前线程获取返回值。 Mono<String> resp = WebClient.create() 		.post() 		.uri("http://localhost:8080/demo/json") 		.contentType(MediaType.APPLICATION_JSON_UTF8) 		.body(BodyInserters.fromObject("{\n" + 				"  \"title\" : \"this is title\",\n" + 				"  \"author\" : \"this is author\"\n" + 				"}")) 		.retrieve().bodyToMono(String.class); System.out.println("---resp.block(): "+resp.block());   // 二进制上传文件 HttpHeaders headers = new HttpHeaders(); headers.setContentType(MediaType.IMAGE_PNG); HttpEntity<ClassPathResource> entity = new HttpEntity<>(new ClassPathResource("parallel.png"), headers); MultiValueMap<String, Object> parts = new LinkedMultiValueMap<>(); parts.add("file", entity); Mono<String> resp = WebClient.create() 		.post() 		.uri("http://localhost:8080/upload") 		.contentType(MediaType.MULTIPART_FORM_DATA) 		.body(BodyInserters.fromMultipartData(parts)) 		.retrieve().bodyToMono(String.class); System.out.println("---resp.block(): "+resp.block());

2.5、WebSocketClient  使用 Socket

WebSocketClient client = new ReactorNettyWebSocketClient(); URI url = new URI("ws://localhost:8080/path"); client.execute(url, session ->session.receive().doOnNext(System.out::println).then());

三、封装工具类 

import org.springframework.http.HttpHeaders; import org.springframework.http.MediaType; import org.springframework.web.reactive.function.BodyInserters; import org.springframework.web.reactive.function.client.WebClient; import reactor.core.publisher.Mono;  /**  *   */ public class WebClientUtils {     	     private WebClient webClient; 	     public WebClientUtils(String baseUrl) {         this.webClient = WebClient.builder()                 .baseUrl(baseUrl)                 .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE) 				.filter(logRequest())                 .build();     } 	     public <T> Mono<T> get(String uri, Class<T> responseType) {         return webClient.get()                 .uri(uri)                 .retrieve()                 .bodyToMono(responseType);     } 	     public <T> Mono<T> post(String uri, Object request, Class<T> responseType) {         return webClient.post()                 .uri(uri)                 .body(BodyInserters.fromValue(request))                 .retrieve()                 .bodyToMono(responseType);     } 	     public <T> Mono<T> put(String uri, Object request, Class<T> responseType) {         return webClient.put()                 .uri(uri)                 .body(BodyInserters.fromValue(request))                 .retrieve()                 .bodyToMono(responseType);     } 	     public <T> Mono<T> delete(String uri, Class<T> responseType) {         return webClient.delete()                 .uri(uri)                 .retrieve()                 .bodyToMono(responseType);     } 	 	private ExchangeFilterFunction logRequest() { 	     return (clientRequest, next) -> {         logger.info("Request: {} {}", clientRequest.method(), clientRequest.url()); 		clientRequest.headers().forEach((name, values) -> values.forEach(value -> logger.info("{}={}", name, value)));         return next.exchange(clientRequest); 		}; 	}  }   public class Test001 {     public static void main(String[] args) {         WebClientUtils webClientUtils = new WebClientUtils("https://api.example.com"); 		         // 发起 GET 请求         webClientUtils.get("/users/1", User.class).subscribe(user -> System.out.println("GET response: " + user)); 		         // 发起 POST 请求         User newUser = new User("John", "Doe");         webClientUtils.post("/users", newUser, User.class).subscribe(user -> System.out.println("POST response: " + user)); 		         // 发起 PUT 请求         User updatedUser = new User("Jane", "Doe");         webClientUtils.put("/users/1", updatedUser, User.class).subscribe(user -> System.out.println("PUT response: " + user)); 		         // 发起 DELETE 请求         webClientUtils.delete("/users/1", Void.class).subscribe(response -> System.out.println("DELETE response: " + response));     } }