SpringBoot遠(yuǎn)程調(diào)用(HTTP)實(shí)現(xiàn)過(guò)程
更新時(shí)間：2025年11月14日 09:52:39 作者：tomorrow.hello
本文介紹了JDK中常用的HTTP客戶端庫(kù),包括HttpURLConnection、Apache HttpClient、OkHttp、RestTemplate和WebClient,并詳細(xì)比較了WebClient和RestTemplate的異同,同時(shí),文章還介紹了Reactor項(xiàng)目中的Mono和Flux等反應(yīng)式編程概念,以及如何在異步非阻塞環(huán)境中使用這些工具
1.JDK中的HttpURLConnection
使用JDK 1.8中自帶的rt.jar包中的java.net中的HttpURLConnection
public static void postTest() throws Exception{
// 1.請(qǐng)求URL
String postUrl = "";
// 2.請(qǐng)求參數(shù)JSON格式
Map<String, String> parammap = new HashMap<>();;
String json = JSON.toJSONString(map);
// 3.創(chuàng)建連接與設(shè)置連接參數(shù)
URL urlObj = new URL(postUrl);
HttpURLConnection httpConn = (HttpURLConnection) urlObj.openConnection();
httpConn.setRequestMethod("POST");
httpConn.setRequestProperty("Charset", "UTF-8");
// POST請(qǐng)求且JSON數(shù)據(jù),必須設(shè)置
httpConn.setRequestProperty("Content-Type", "application/json");
// 打開(kāi)輸出流,默認(rèn)是false
httpConn.setDoOutput(true);
// 打開(kāi)輸入流,默認(rèn)是true,可省略
httpConn.setDoInput(true);
// 4.從HttpURLConnection獲取輸出流和寫(xiě)數(shù)據(jù)
OutputStream oStream = httpConn.getOutputStream();
oStream.write(json.getBytes());
oStream.flush();
// 5.發(fā)起http調(diào)用(getInputStream觸發(fā)http請(qǐng)求)
if (httpConn.getResponseCode() != 200) {
throw new Exception("調(diào)用服務(wù)端異常.");
}
// 6.從HttpURLConnection獲取輸入流和讀數(shù)據(jù)
BufferedReader br = new BufferedReader(
new InputStreamReader(httpConn.getInputStream()));
String resultData = br.readLine();
System.out.println("從服務(wù)端返回結(jié)果: " + resultData);
// 7.關(guān)閉HttpURLConnection連接
httpConn.disconnect();
}2.commons-httpclient中的HttpClient
引入依賴
<dependency> <groupId>commons-httpclient</groupId> <artifactId>commons-httpclient</artifactId> </dependency>
public static void postTest() throws Exception {
// 1.請(qǐng)求URL
String postUrl = "";
// 2.請(qǐng)求參數(shù)
Map<String, String> parammap = new HashMap<>();
String json = JSON.toJSONString(parammap);
// 3.創(chuàng)建連接與設(shè)置連接參數(shù)
HttpClient httpClient = new HttpClient();
PostMethod postMethod = new PostMethod(postUrl);
postMethod.addRequestHeader("Content-Type", "application/json");
RequestEntity entity = new StringRequestEntity(json, "application/json", "UTF-8");
postMethod.setRequestEntity(entity);
//解決返回值中文亂碼
postMethod.getParams().setParameter(HttpMethodParams.HTTP_CONTENT_CHARSET, "UTF-8");
String resultData = "";
// 4.發(fā)起請(qǐng)求
int code = httpClient.executeMethod(postMethod);
if (code != 200) {
throw new Exception("調(diào)用服務(wù)端異常.");
}
// 5.接收返回值
resultData = postMethod.getResponseBodyAsString();
System.out.println("從服務(wù)端返回結(jié)果: " + resultData);
// 6.關(guān)閉連接
postMethod.releaseConnection();
}3.httpclient中的HttpClientBuilder
HttpClient:是apache httpClient包下的，代碼復(fù)雜，需要資源回收。
引入依賴
<dependency> <groupId>org.apache.httpcomponents</groupId> <artifactId>httpclient</artifactId> </dependency>
public static void postTest() throws Exception {
// 1.請(qǐng)求URL
String postUrl = "";
// 2.請(qǐng)求參數(shù)
Map<String, String> paramMap = new HashMap<>();
String json = JSON.toJSONString(paramMap);
// 3.創(chuàng)建連接與設(shè)置連接參數(shù)
CloseableHttpClient httpClient = HttpClientBuilder.create().build();
HttpPost httpPost = new HttpPost(postUrl);
StringEntity entity = new StringEntity(json);
entity.setContentEncoding("UTF-8");
entity.setContentType("application/json");
httpPost.setEntity(entity);
// 4.發(fā)起請(qǐng)求與接收返回值
HttpResponse response = httpClient.execute(httpPost);
if (response.getStatusLine().getStatusCode() != 200) {
throw new Exception("調(diào)用服務(wù)端異常.");
}
HttpEntity res = response.getEntity();
String resultData = EntityUtils.toString(res);
System.out.println("從服務(wù)端返回結(jié)果: " + resultData);
// 5.關(guān)閉連接
httpClient.close();
}4.okhttp中的OkHttpClient
引入依賴
<dependency>
<groupId>com.squareup.okhttp3</groupId>
<artifactId>okhttp</artifactId>
<exclusions>
<exclusion>
<groupId>com.google.android</groupId>
<artifactId>android</artifactId>
</exclusion>
</exclusions>
</dependency>public static void postTest() throws Exception {
// 1.請(qǐng)求URL
String postUrl = "";
// 2.請(qǐng)求參數(shù)
Map<String, String> paramMap = new HashMap<>();
String json = JSON.toJSONString(paramMap);
// 3.創(chuàng)建連接與設(shè)置連接參數(shù)
MediaType mediaType = MediaType.parse("application/json; charset=UTF-8");
RequestBody requestBody = RequestBody.Companion.create(json, mediaType);
Request request = new Request.Builder().url(postUrl).post(requestBody).build();
OkHttpClient okHttpClient = new OkHttpClient.Builder().build();
// 4.發(fā)起請(qǐng)求與接收返回值
Response response = okHttpClient.newCall(request).execute();
String resultData = response.body().string();
System.out.println("從服務(wù)端返回結(jié)果: " + resultData);
}5.RestClient
官網(wǎng)解釋： synchronous client with a fluent API.
5.RestTemplate（WebMVC推薦）
RestTemplate：RestTemplate 是 Spring 提供的一個(gè)經(jīng)典同步 HTTP 客戶端工具，可以用于調(diào)用 RESTful 風(fēng)格的外部接口，代碼簡(jiǎn)單，默認(rèn)依賴jdk的HTTP連接工具，它是一個(gè)同步阻塞型客戶端，官網(wǎng)解釋：synchronous client with template method API.
RestTemplate restTemplate = new RestTemplate();
String url = "http://api.example.com/user/{id}";
Map<String, String> params = new HashMap<>();
params.put("id", "123");
User user = restTemplate.getForObject(url, User.class, params);
使用 RestTemplate并發(fā)調(diào)用：
public class RestTemplateConcurrentExample {
private RestTemplate restTemplate = new RestTemplate();
public void fetchMultipleUsers(String[] userIds) {
ExecutorService executor = Executors.newFixedThreadPool(userIds.length);
for (String userId : userIds) {
executor.submit(() -> {
String url = "https://api.example.com/users/" + userId;
String response = restTemplate.getForObject(url, String.class);
System.out.println(response);
});
}
executor.shutdown();
}
}6.WebClient（WebFlux推薦）
WebClient 是 Spring 5 引入的一種非阻塞式、響應(yīng)式的 HTTP 客戶端工具，它提供了一套簡(jiǎn)潔的 API 來(lái)發(fā)送 HTTP 請(qǐng)求并處理響應(yīng)。WebClient 基于 Reactor 提供了對(duì)響應(yīng)式編程的支持，可以實(shí)現(xiàn)高性能的異步操作，官網(wǎng)解釋：non-blocking, reactive client with fluent API。
1.簡(jiǎn)單使用
WebClient webClient = WebClient.create();
String url = "http://api.example.com/user/{id}";
Map<String, String> params = new HashMap<>();
params.put("id", "123");
Mono<User> result = webClient.get()
.uri(uriBuilder -> uriBuilder.path(url).build(params))
.retrieve()
.bodyToMono(User.class);
result.subscribe(user -> {
// 處理響應(yīng)結(jié)果
});webclient并發(fā)調(diào)用:
public class WebClientConcurrentExample {
private WebClient webClient = WebClient.create();
public Flux<String> fetchMultipleUsers(String[] userIds) {
return Flux.fromArray(userIds)
.flatMap(userId -> webClient.get()
.uri("https://api.example.com/users/" + userId)
.retrieve()
.bodyToMono(String.class));
}
}2.高級(jí)使用
1.配置連接池
/**
WebClient連接池
**/
@Configuration
public class WebClientConfig {
@Bean
public WebClient webClient() {
// 配置HTTP連接池
ConnectionProvider provider = ConnectionProvider.builder("custom")
.maxConnections(500)
.maxIdleTime(Duration.ofSeconds(20))
.build();
// 配置HTTP客戶端
HttpClient httpClient = HttpClient.create(provider)
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000)
.responseTimeout(Duration.ofSeconds(5))
.doOnConnected(conn ->
conn.addHandlerLast(new ReadTimeoutHandler(5))
.addHandlerLast(new WriteTimeoutHandler(5)));
// 構(gòu)建WebClient實(shí)例
return WebClient.builder()
.clientConnector(new ReactorClientHttpConnector(httpClient))
.baseUrl("https://echo.apifox.com")
.defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
.defaultHeader(HttpHeaders.ACCEPT, MediaType.APPLICATION_JSON_VALUE)
// 添加請(qǐng)求日志記錄功能
.filter(ExchangeFilterFunction.ofRequestProcessor(
clientRequest -> {
log.debug("Request: {} {}",
clientRequest.method(),
clientRequest.url());
return Mono.just(clientRequest);
}
))
// 添加響應(yīng)日志記錄功能
.filter(ExchangeFilterFunction.ofResponseProcessor(
clientResponse -> {
log.debug("Response status: {}",
clientResponse.statusCode());
return Mono.just(clientResponse);
}
))
.build();
}
}2. retrieve()和exchange()區(qū)別
retrieve()
- 用途：retrieve() 方法用于簡(jiǎn)化響應(yīng)處理，特別是當(dāng)你只需要響應(yīng)體時(shí)。
- 自動(dòng)錯(cuò)誤處理：retrieve() 會(huì)自動(dòng)處理 HTTP 錯(cuò)誤狀態(tài)碼（例如 4xx 和 5xx），并拋出 WebClientResponseException 及其子類。
- 返回值：通常用于直接獲取響應(yīng)體，例如 bodyToMono(String.class) 或 bodyToFlux(String.class)。
- 適用場(chǎng)景：適用于大多數(shù)常見(jiàn)的請(qǐng)求處理場(chǎng)景，特別是當(dāng)你不需要手動(dòng)處理響應(yīng)狀態(tài)碼時(shí)。
public Mono<JSONObject> get(String q1) {
return webClient.get()
.uri(uriBuilder -> uriBuilder
.path("/get")
.queryParam("q1", q1)
.build())
.accept(MediaType.APPLICATION_JSON)
.retrieve()
.bodyToMono(JSONObject.class);
}exchange()
- 用途：exchange() 方法提供了更底層的控制，允許你手動(dòng)處理響應(yīng)，包括響應(yīng)狀態(tài)碼和響應(yīng)頭。
- 手動(dòng)錯(cuò)誤處理：exchange() 不會(huì)自動(dòng)處理 HTTP 錯(cuò)誤狀態(tài)碼，你需要手動(dòng)檢查響應(yīng)狀態(tài)碼并進(jìn)行相應(yīng)的處理。
- 返回值：返回 ClientResponse 對(duì)象，你可以從中提取響應(yīng)狀態(tài)碼、響應(yīng)頭和響應(yīng)體。
- 適用場(chǎng)景：適用于需要手動(dòng)處理響應(yīng)狀態(tài)碼或響應(yīng)頭的復(fù)雜場(chǎng)景。
public Mono<JSONObject> get(String q1) {
return webClient.get()
.uri(uriBuilder -> uriBuilder
.path("/get")
.queryParam("q1", q1)
.build())
.accept(MediaType.APPLICATION_JSON)
.exchangeToMono(response -> {
if (response.statusCode().is2xxSuccessful()) {
return response.bodyToMono(JSONObject.class);
} else {
return Mono.error(new RuntimeException("Request failed with status code: " + response.statusCode()));
}
});
}3. GET，POST，PUT，DELETE請(qǐng)求
@Service
public class ApiService {
@Resource
private WebClient webClient;
// GET請(qǐng)求
public Mono<JSONObject> get(String q1) {
return webClient.get()
.uri(uriBuilder -> uriBuilder
.path("/get")
.queryParam("q1", q1)
.build())
.accept(MediaType.APPLICATION_JSON)
.retrieve()
.bodyToMono(JSONObject.class);
}
// POST請(qǐng)求
public Mono<JSONObject> post(JSONObject body) {
return webClient.post()
.uri("/post")
.bodyValue(body)
.retrieve()
.bodyToMono(JSONObject.class);
}
// PUT請(qǐng)求
public Mono<JSONObject> put(String q1, JSONObject JSONObject) {
return webClient.put()
.uri(uriBuilder -> uriBuilder
.path("/put")
.queryParam("q1", q1)
.build())
.bodyValue(JSONObject)
.retrieve()
.bodyToMono(JSONObject.class);
}
// DELETE請(qǐng)求
public Mono<JSONObject> delete(String q1) {
return webClient.delete()
.uri(uriBuilder -> uriBuilder
.path("/delete")
.queryParam("q1", q1)
.build())
.retrieve()
.bodyToMono(JSONObject.class);
}
}3.Mono和Flux
Mono用于表示包含 0 個(gè)或 1 個(gè)元素的異步數(shù)據(jù)流（如單個(gè)用戶信息查詢結(jié)果）；Flux用于表示包含 0 個(gè)或多個(gè)元素的異步數(shù)據(jù)流（如列表查詢結(jié)果）。
4. 創(chuàng)建型操作符(just)
1. Mono和Flux
just操作可直接通過(guò)傳入的元素創(chuàng)建一個(gè)響應(yīng)式流。其中，Flux 可接收多個(gè)元素（0 個(gè)或多個(gè)），Mono 則僅能接收單個(gè)元素（若傳入多個(gè)會(huì)報(bào)錯(cuò)，示例中Mono.just("A", "B", "C")為錯(cuò)誤示范，正確應(yīng)為Mono.just("A")）。
// 創(chuàng)建包含3個(gè)元素的Flux流（可正常運(yùn)行）
Flux<String> flux = Flux.just("A", "B", "C");
// 正確示例：創(chuàng)建僅包含1個(gè)元素的Mono流
Mono<String> mono = Mono.just("A");
2. fromIterable(Iterable<T>)
fromIterable、fromArray則分別將已有的Iterable類型集合（如List、Set），和數(shù)組轉(zhuǎn)換為響應(yīng)式流，自動(dòng)遍歷集合中的元素并發(fā)送到流中。適合需要處理已有數(shù)據(jù)集合的場(chǎng)景，避免手動(dòng)逐個(gè)添加元素。
// fromIterable
List<Integer> list = Arrays.asList(1, 2, 3);
Flux<Integer> flux = Flux.fromIterable(list);
// fromArray
Integer[] arr = {10, 20, 30};
Flux<Integer> flux = Flux.fromArray(arr);
flux.subscribe(System.out::println);
3. range(int start, int count)
range則適用于快速生成一段連續(xù)的整數(shù)流來(lái)構(gòu)建測(cè)試用例。其中，第一個(gè)參數(shù)為起始值，第二個(gè)參數(shù)為元素個(gè)數(shù)。
Flux.range(1, 5).subscribe(System.out::println); // 輸出 1~5
4. create(FluxSink<T>)
通過(guò)FluxSink對(duì)象可手動(dòng)發(fā)送元素（next）、結(jié)束信號(hào)（complete）或錯(cuò)誤信號(hào)（error），靈活控制流的產(chǎn)生過(guò)程。
該API適合用于從異步回調(diào)、事件監(jiān)聽(tīng)中獲取數(shù)據(jù)等場(chǎng)景。
// 手動(dòng)創(chuàng)建流，通過(guò)sink發(fā)送元素并結(jié)束
Flux.create(sink -> {
sink.next("Hello"); // 發(fā)送第一個(gè)元素
sink.next("WebFlux"); // 發(fā)送第二個(gè)元素
sink.complete(); // 標(biāo)記流結(jié)束（不再發(fā)送元素）
}).subscribe(System.out::println); // 輸出：Hello WebFlux
Mono<Integer> mono = Mono.create(monoSink -> {
// 成功的信號(hào)
//monoSink.success(111);
// 異常的信號(hào)
monoSink.error(new Exception("this is error"));
});
mono.onErrorResume(Exception.class, ex -> {
System.out.println("error:" + ex);
return Mono.just(12345);
}).subscribe(rr -> {
System.out.println("result:" + rr);
});5. 轉(zhuǎn)換型操作符(map)
1. map(Function<T, R>)
map的操作則主要對(duì)流中的每個(gè)元素執(zhí)行指定的轉(zhuǎn)換操作。即輸入一個(gè)元素，輸出一個(gè)轉(zhuǎn)換后的元素，保持流的元素?cái)?shù)量不變。適合簡(jiǎn)單的同步轉(zhuǎn)換場(chǎng)景。
// 對(duì)流中每個(gè)字符串執(zhí)行"轉(zhuǎn)大寫(xiě)"操作
Flux<String> flux = Flux.just("apple", "banana")
.map(String::toUpperCase); // 調(diào)用String的toUpperCase()方法
flux.subscribe(System.out::println); // 輸出：APPLE BANANA
2. flatMap(Function<T, Publisher<R>>)
與map不同，flatMap 則接收一個(gè)元素T，返回一個(gè)新的響應(yīng)式流Publisher<R>（如Flux<R>或Mono<R>）。 即整個(gè)過(guò)程是 "元素→流" 的映射，會(huì)將子流 "扁平化" 合并為一個(gè)新流，輸出流的元素?cái)?shù)量可能比輸入流多（或少）。
// 將每個(gè)字符串按字符拆分，轉(zhuǎn)換為包含單個(gè)字符的子流，再合并
Flux<String> flux = Flux.just("hello", "world")
.flatMap(s -> Flux.fromArray(s.split(""))); // 拆分后子流為 ["h","e","l","l","o"] 和 ["w","o","r","l","d"]
flux.subscribe(System.out::println);
// 可能輸出：h w e o l r l l d（順序不固定，因兩個(gè)子流并行處理）
舉個(gè)更實(shí)際的例子，如 "一個(gè)訂單包含多個(gè)商品，需要根據(jù)訂單 ID 查詢所有商品" 的場(chǎng)景（1 個(gè)訂單→多個(gè)商品組成的流），或需要在轉(zhuǎn)換中調(diào)用異步操作可使用flatMap來(lái)進(jìn)行操作。
// 模擬"根據(jù)用戶ID查詢多個(gè)訂單"的異步操作
Flux.just(1001, 1002) // 用戶ID流
.flatMap(userId -> orderService.findOrdersByUserId(userId)) // 每個(gè)用戶ID→訂單流
.subscribe(order -> System.out.println("訂單：" + order));
3. concatMap(Function<T, Publisher<R>>)
與flatMap類似，concatMap則是將每個(gè)元素轉(zhuǎn)換為子流后合并，但嚴(yán)格按照原元素的順序處理子流（前一個(gè)子流完全處理完才會(huì)處理下一個(gè)），因此最終流的元素順序與原元素對(duì)應(yīng)的子流順序一致。適合需要保證順序的場(chǎng)景（如按順序處理批量任務(wù)）。
// 同樣拆分字符串為字符子流，但按原順序合并
Flux<String> flux = Flux.just("hello", "world")
.concatMap(s -> Flux.fromArray(s.split(""))); // 先處理"hello"的子流，再處理"world"的子流
flux.subscribe(System.out::println);
// 固定輸出：h e l l o w o r l d（嚴(yán)格遵循原元素順序）
6 過(guò)濾型操作符
1. filter(Predicate<T>)
Flux<Integer> flux = Flux.range(1, 10).filter(i -> i % 2 == 0); flux.subscribe(System.out::println);
2. distinct()
對(duì)流中所有元素進(jìn)行去重處理，保留首次出現(xiàn)的元素，后續(xù)重復(fù)元素會(huì)被過(guò)濾。
Flux.just(1, 2, 2, 3).distinct().subscribe(System.out::println);
3. limitRate(int rate)
控制流從上游數(shù)據(jù)源獲取元素的速率，每次向上游請(qǐng)求 rate 個(gè)元素，處理完后再請(qǐng)求下一批，避免一次性加載過(guò)多數(shù)據(jù)導(dǎo)致內(nèi)存壓力（類似 “分批拉取”）。常用于流中元素?cái)?shù)量極大的場(chǎng)景（如處理百萬(wàn)級(jí)數(shù)據(jù)），平衡內(nèi)存占用與處理效率。
// 生成1~100的整數(shù)流，每次從上游獲取10個(gè)元素后再繼續(xù)請(qǐng)求 Flux.range(1, 100) .limitRate(10).subscribe(System.out::println); // 內(nèi)部過(guò)程：先請(qǐng)求1~10，處理完后再請(qǐng)求11~20，直到所有元素處理完畢
7.異常處理
- onStatus：用于處理HTTP響應(yīng)狀態(tài)碼，允許根據(jù)不同狀態(tài)碼執(zhí)行特定邏輯。例如，當(dāng)收到404狀態(tài)碼時(shí)重定向到錯(cuò)誤頁(yè)面，或根據(jù)200狀態(tài)碼正常處理數(shù)據(jù)。
- onErrorResume：用于處理異常情況（如超時(shí)、連接失敗等），切換到一個(gè)備用的Publisher。。例如，當(dāng)請(qǐng)求超時(shí)時(shí)返回默認(rèn)數(shù)據(jù)或重試請(qǐng)求。
- onErrorReturn：當(dāng)發(fā)生錯(cuò)誤時(shí)，返回一個(gè)默認(rèn)值。
- onErrorMap：將發(fā)生的錯(cuò)誤轉(zhuǎn)換為另一種類型的錯(cuò)誤。
- doOnError：在發(fā)生錯(cuò)誤時(shí)執(zhí)行一些操作，但不改變流本身。
核心差異：
- 觸發(fā)時(shí)機(jī)：onStatus在響應(yīng)階段觸發(fā)，onErrorResume在請(qǐng)求階段觸發(fā)。
- 處理范圍：onStatus僅處理HTTP狀態(tài)碼異常，onErrorResume處理所有請(qǐng)求異常（包括網(wǎng)絡(luò)問(wèn)題、超時(shí)等）。
- 鏈?zhǔn)讲僮?zwnj;：onErrorResume支持鏈?zhǔn)秸{(diào)用（如重試、返回默認(rèn)值等），onStatus通常僅用于分支處理。
// 處理錯(cuò)誤響應(yīng)
public Mono<JSONObject> getUserWithErrorHandling(Long id) {
return webClient.get()
.uri("/users/{id}", id)
.retrieve()
.onStatus(HttpStatusCode::is4xxClientError, clientResponse -> Mono.error(new RuntimeException("客戶端錯(cuò)誤")))
.onStatus(HttpStatusCode::is5xxServerError, clientResponse -> Mono.error(new RuntimeException("服務(wù)器錯(cuò)誤")))
.bodyToMono(JSONObject.class);
} Mono<ResponseEntity<String>> mono = WebClient.create().get().uri("http://")
.retrieve()
.toEntity(String.class)
.onErrorResume(WebClientResponseException.class, e -> {
if (e.getStatusCode().is4xxClientError()) {
return Mono.error(new HttpClientErrorException(e.getStatusCode(), e.getResponseBodyAsString()));
}
return Mono.error(e);
});8. 時(shí)間控制操作符
1. delayElements(Duration duration)
delayElements主要讓流中的每個(gè)元素都延遲指定時(shí)間后再發(fā)射，相當(dāng)于給每個(gè)元素的發(fā)送增加一個(gè)統(tǒng)一的 “等待期”。
// 生成1~3的整數(shù)流，每個(gè)元素延遲1秒后發(fā)送
Flux.range(1, 3).delayElements(Duration.ofSeconds(1))
.subscribe(System.out::println); // 依次間隔1秒輸出：1、2、3
Thread.sleep(5000); // 主線程休眠5秒，防止程序提前退出（否則看不到完整輸出）
2. timeout(Duration timeout)
timeout主要為流設(shè)置超時(shí)閾值，若流在指定時(shí)間內(nèi)沒(méi)有發(fā)射新元素（或未完成），則會(huì)觸發(fā)超時(shí)錯(cuò)誤（TimeoutException）。適合需要限制操作響應(yīng)時(shí)間的場(chǎng)景（如接口調(diào)用超時(shí)控制）。
// 生成1~3的整數(shù)流，每個(gè)元素延遲500毫秒發(fā)送，同時(shí)設(shè)置超時(shí)時(shí)間為300毫秒
Flux.range(1, 3)
.delayElements(Duration.ofMillis(500)) // 元素發(fā)送間隔500ms
.timeout(Duration.ofMillis(300)) // 超過(guò)300ms未收到新元素則超時(shí)
.subscribe(
System.out::println, // 正常元素處理（此處第一個(gè)元素就會(huì)超時(shí)，不會(huì)執(zhí)行）
Throwable::printStackTrace // 捕獲并打印超時(shí)異常
);
Thread.sleep(2000); // 主線程休眠，確保異常能被捕獲
// 輸出：TimeoutException（因第一個(gè)元素需500ms發(fā)送，超過(guò)300ms超時(shí)閾值）9 訂閱操作符
1. subscribe(Consumer<T>)
Mono的subscribe方法用于觸發(fā)異步操作鏈的執(zhí)行，其核心機(jī)制涉及發(fā)布-訂閱關(guān)系鏈的構(gòu)建與執(zhí)行。
Mono.just("hello")
.filter(t -> t.startsWith("h"))
.map(String::toUpperCase)
.subscribe(System.out::println);
#此代碼通過(guò)subscribe觸發(fā)處理鏈，依次執(zhí)行過(guò)濾、轉(zhuǎn)換操作。2. doOnNext(Consumer<T>)
在流中的每個(gè)元素被發(fā)射到訂閱者之前，觸發(fā)指定的消費(fèi)函數(shù)（如日志記錄、數(shù)據(jù)預(yù)處理等），但不會(huì)改變?cè)乇旧砘蛄鞯慕Y(jié)構(gòu)。
// 生成1~3的整數(shù)流，發(fā)射前打印提示，再將元素發(fā)送給訂閱者
Flux.range(1, 3)
.doOnNext(i -> System.out.println("Before emit: " + i)) // 發(fā)射前執(zhí)行：打印提示
.subscribe(System.out::println); // 訂閱者接收并打印元素
// 輸出：
// Before emit: 1 → 發(fā)射前操作
// 1 → 訂閱者處理
// Before emit: 2
// 2
// Before emit: 3
// 3
3. doOnComplete(Runnable)
流正常結(jié)束（所有元素發(fā)射完畢且無(wú)錯(cuò)誤）時(shí)，觸發(fā)指定的無(wú)參任務(wù)（Runnable），可用于執(zhí)行流結(jié)束后的收尾操作（如釋放資源、打印完成日志等）。
// 創(chuàng)建包含1、2的流，完成時(shí)打印"Done"，并訂閱（無(wú)需處理元素，僅觸發(fā)完成回調(diào)）
Flux.just(1, 2)
.doOnComplete(() -> System.out.println("Done")) // 流正常結(jié)束時(shí)執(zhí)行
.subscribe(); // 訂閱啟動(dòng)流
// 輸出：Done（當(dāng)1和2都發(fā)射完畢后，觸發(fā)完成回調(diào)）10 多個(gè)Mono的處理
1. 合并多個(gè)Mono
如果你有多個(gè)獨(dú)立的Mono實(shí)例，并希望它們?cè)谕粋€(gè)上下文中執(zhí)行（例如，并行執(zhí)行后合并結(jié)果），你可以使用Mono.zip、Mono.when或Mono.merge等方法。
#使用Mono.zip
Mono<String> mono1 = webClient.get()
.uri("http://example.com/api/data1")
.retrieve()
.bodyToMono(String.class);
Mono<Integer> mono2 = webClient.get()
.uri("http://example.com/api/data2")
.retrieve()
.bodyToMono(Integer.class);
Mono<Tuple2<String, Integer>> combined = Mono.zip(mono1, mono2);
#使用Mono.when
Mono<Void> mono1 = webClient.get()
.uri("http://example.com/api/data1")
.retrieve()
.bodyToMono(String.class)
.then(); // 轉(zhuǎn)換為Void Mono
Mono<Void> mono2 = webClient.get()
.uri("http://example.com/api/data2")
.retrieve()
.bodyToMono(Integer.class)
.then(); // 轉(zhuǎn)換為Void Mono
Mono<Void> combined = Mono.when(mono1, mono2);2 順序執(zhí)行多個(gè)Mono
如果你需要按順序執(zhí)行多個(gè)Mono，可以使用.then或Mono.concat。
#使用Mono.flatMap
Mono<String> mono1 = webClient.get()
.uri("http://example.com/api/data1")
.retrieve()
.bodyToMono(String.class);
Mono<Integer> mono2 = mono1 // 從mono1的結(jié)果繼續(xù)執(zhí)行
.flatMap(result -> webClient.get()
.uri("http://example.com/api/data2?param=" + result)
.retrieve()
.bodyToMono(Integer.class));
#使用Mono.concat
Flux<Object> fluxOfMonos = Flux.just(
webClient.get().uri("http://example.com/api/data1").retrieve().bodyToMono(String.class),
webClient.get().uri("http://example.com/api/data2").retrieve().bodyToMono(Integer.class)
);
Flux<Object> result = Flux.concat(fluxOfMonos); // 注意：這會(huì)按順序合并結(jié)果，但保持它們?yōu)镸onos，而非合并成一個(gè)結(jié)果。3 轉(zhuǎn)換多個(gè)Mono為列表或其他集合類型
如果你想要將多個(gè)Mono的結(jié)果收集到一個(gè)列表或其他集合中，可以使用Flux.fromArray或Flux.fromIterable與.collectList()或.collectMap()等操作。
Flux<Object> fluxOfMonos = Flux.fromArray(new Object[]{
webClient.get().uri("http://example.com/api/data1").retrieve().bodyToMono(String.class),
webClient.get().uri("http://example.com/api/data2").retrieve().bodyToMono(Integer.class)
});
Mono<List<Object>> resultList = fluxOfMonos.collectList(); // 收集為L(zhǎng)ist11. block()/blockFirst()/blockLast()
因?yàn)閎lock是阻塞操作，所以不能在webflux框架中阻塞任務(wù)結(jié)果，因?yàn)閞eactor-netty屬于異步非阻塞框架。在異步非阻塞中的線程中阻塞結(jié)果會(huì)報(bào)錯(cuò)。
正確的做法新建一個(gè)線程阻塞結(jié)果
12. 使用mono后臺(tái)執(zhí)行
在Java中，特別是在使用Spring框架時(shí)，我們經(jīng)常需要異步執(zhí)行代碼以避免阻塞主線程。Spring 5引入了Project Reactor，它提供了Mono和Flux等反應(yīng)式類型，使得編寫(xiě)非阻塞代碼變得非常方便。
1. Mono.fromRunnable
@GetMapping("/backstagecall")
public Mono<String> backstageCall() {
Mono<String> reulst = Mono.just("backgraudCall call");
// 方法1：使用Mono.fromRunnable
Mono.fromRunnable(() -> {
log.info("backstageMono start ");
// 耗時(shí)的代碼
mysleep(10);
log.info("backstageMono end ");
}).subscribeOn(Schedulers.boundedElastic()) // 在后臺(tái)執(zhí)行
.subscribe();
log.info("backstagecall end");
return reulst;
}
2. Mono.defer+Schedulers
@GetMapping("/backstagecall")
public Mono<String> backstageCall() {
Mono<String> reulst = Mono.just("backgraudCall call");
// 方法2：使用Mono.defer+Schedulers
Mono.defer(() -> {
log.info("backstageMono start ");
mysleep(10);
log.info("backstageMono end ");
return Mono.empty();
}).subscribeOn(Schedulers.boundedElastic()) // 在后臺(tái)執(zhí)行
.subscribe();
log.info("backstagecall end");
return reulst;
}
3. Mono.create
@GetMapping("/backstagecall")
public Mono<String> backstageCall() {
Mono<String> reulst = Mono.just("backgraudCall call");
// 方法3：使用Mono.create
Mono.create(sink -> {
Schedulers.boundedElastic().schedule(() -> {
log.info("backstageMono start ");
mysleep(10);
log.info("backstageMono end ");
// 完成信號(hào)
sink.success("111");
});
}).subscribe();
log.info("backstagecall end");
return reulst;
}
7.OpenFeign或者Feign
Feign 是 Spring Cloud 提供的一個(gè)聲明式的 HTTP 客戶端工具，它基于注解和接口定義的方式，簡(jiǎn)化了外部接口調(diào)用的流程。Feign 集成了 Ribbon 負(fù)載均衡和 Hystrix 熔斷器等功能，使得接口調(diào)用更加靈活可靠。
@FeignClient(name = "user-service", url = "http://api.example.com")
public interface UserFeignClient {
@GetMapping("/user/{id}")
User getUser(@PathVariable("id") String id);
}
總結(jié)
WebClient與RestTemplate對(duì)比
特性 | WebClient | RestTemplate |
|---|---|---|
編程模型 |
|
|
性能 | 更好 | 一般 |
資源利用 | 更高效 | 一般 |
學(xué)習(xí)曲線 | 較陡 | 平緩 |
適用場(chǎng)景 | 高并發(fā)、響應(yīng)式系統(tǒng) | 簡(jiǎn)單應(yīng)用、傳統(tǒng)系統(tǒng) |
以上為個(gè)人經(jīng)驗(yàn)，希望能給大家一個(gè)參考，也希望大家多多支持腳本之家。
您可能感興趣的文章:
- SpringBoot基于AOP的本地/遠(yuǎn)程調(diào)用動(dòng)態(tài)路由實(shí)踐指南
- SpringBoot項(xiàng)目中使用Netty實(shí)現(xiàn)遠(yuǎn)程調(diào)用的示例代碼
- springboot3整合遠(yuǎn)程調(diào)用的過(guò)程解析
- SpringBoot整合RabbitMQ實(shí)現(xiàn)RPC遠(yuǎn)程調(diào)用功能
- SpringBoot?Http遠(yuǎn)程調(diào)用的方法
- springBoot使用openfeign來(lái)遠(yuǎn)程調(diào)用的實(shí)現(xiàn)
- 在SpringBoot中,如何使用Netty實(shí)現(xiàn)遠(yuǎn)程調(diào)用方法總結(jié)
相關(guān)文章
Java AbstractMethodError案例分析詳解
這篇文章主要介紹了Java AbstractMethodError案例分析詳解,本篇文章通過(guò)簡(jiǎn)要的案例,講解了該項(xiàng)技術(shù)的了解與使用,以下就是詳細(xì)內(nèi)容,需要的朋友可以參考下2021-08-08
一段代碼搞懂關(guān)于Java中List、Set集合及Map的使用
這篇文章主要介紹了關(guān)于Java中List、Set集合及Map的使用及l(fā)ist，set和map三者的區(qū)別介紹，非常不錯(cuò)，具有參考借鑒價(jià)值,需要的朋友可以參考下2016-08-08
很多同學(xué)在日常寫(xiě)Java的時(shí)候很少去關(guān)心性能問(wèn)題，但是在我們寫(xiě)代碼的過(guò)程中必須考慮到性能對(duì)程序的影響。小到我們使用位運(yùn)算來(lái)實(shí)現(xiàn)算術(shù)運(yùn)算，大到我們對(duì) Java 代碼的總體架構(gòu)設(shè)計(jì)，性能其實(shí)離我們很近。本文介紹了Java中性能優(yōu)化的35種方法,需要的朋友可以參考下。2017-01-01
這篇文章主要介紹了Spring MVC 請(qǐng)求處理流程詳解,本文分步驟結(jié)合實(shí)例代碼給大家介紹的非常詳細(xì),感興趣的朋友一起看看吧2025-04-04
本篇文章主要介紹了Java 操作Properties配置文件詳解，詳細(xì)的介紹了Properties和主要方法，有興趣的可以了解下2017-05-05
這篇文章主要給大家介紹了關(guān)于Java中ArrayList集合的常用方法，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2021-01-01
MybatisPlus查詢數(shù)據(jù)日期格式化問(wèn)題解決方法
MyBatisPlus是MyBatis的增強(qiáng)工具,支持常規(guī)的CRUD操作以及復(fù)雜的聯(lián)表查詢等功能,這篇文章主要給大家介紹了關(guān)于MybatisPlus查詢數(shù)據(jù)日期格式化問(wèn)題的解決方法,需要的朋友可以參考下2023-10-10
Java幾個(gè)實(shí)例帶你進(jìn)階升華上篇
與其明天開(kāi)始，不如現(xiàn)在行動(dòng)，本文為你帶來(lái)幾個(gè)Java書(shū)寫(xiě)的實(shí)際案例，對(duì)鞏固編程的基礎(chǔ)能力很有幫助，快來(lái)一起往下看看吧2022-03-03
這篇文章主要介紹了Java多線程volatile原理及用法解析,文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2020-07-07
運(yùn)行在服務(wù)器端的小程序，Servlet就是一個(gè)接口，定義了Java類被瀏覽器訪問(wèn)到(tomcat識(shí)別)的規(guī)則，將來(lái)我們自定義一個(gè)類，實(shí)現(xiàn)Servlet接口，復(fù)寫(xiě)方法2021-09-09