Java11 HttpClient小試牛刀

HttpComponents · 發表 2018-09-29 17:44:25

摘要：從java9的jdk.incubator.httpclient模組遷移到java.net.http模組，包名由jdk.incubator.http改為java.net.http 原來的諸如HttpResponse.BodyHandler.asString()方法變更為Http...

從java9的jdk.incubator.httpclient模組遷移到java.net.http模組，包名由jdk.incubator.http改為java.net.http
原來的諸如HttpResponse.BodyHandler.asString()方法變更為HttpResponse.BodyHandlers.ofString()，變化一為BodyHandler改為BodyHandlers，變化二為asXXX()之類的方法改為ofXXX()，由as改為of

例項

設定超時時間

@Test
public void testTimeout() throws IOException, InterruptedException {
//1.set connect timeout
HttpClient client = HttpClient.newBuilder()
.connectTimeout(Duration.ofMillis(5000))
.followRedirects(HttpClient.Redirect.NORMAL)
.build();

//2.set read timeout
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create("http://openjdk.java.net/"))
.timeout(Duration.ofMillis(5009))
.build();

HttpResponse<String> response =
client.send(request, HttpResponse.BodyHandlers.ofString());

System.out.println(response.body());

}
複製程式碼

HttpConnectTimeoutException例項

Caused by: java.net.http.HttpConnectTimeoutException: HTTP connect timed out
at java.net.http/jdk.internal.net.http.ResponseTimerEvent.handle(ResponseTimerEvent.java:68)
at java.net.http/jdk.internal.net.http.HttpClientImpl.purgeTimeoutsAndReturnNextDeadline(HttpClientImpl.java:1248)
at java.net.http/jdk.internal.net.http.HttpClientImpl$SelectorManager.run(HttpClientImpl.java:877)
Caused by: java.net.ConnectException: HTTP connect timed out
at java.net.http/jdk.internal.net.http.ResponseTimerEvent.handle(ResponseTimerEvent.java:69)
... 2 more
複製程式碼

HttpTimeoutException例項

java.net.http.HttpTimeoutException: request timed out

at java.net.http/jdk.internal.net.http.HttpClientImpl.send(HttpClientImpl.java:559)
at java.net.http/jdk.internal.net.http.HttpClientFacade.send(HttpClientFacade.java:119)
at com.example.HttpClientTest.testTimeout(HttpClientTest.java:40)
複製程式碼

設定authenticator

@Test
public void testBasicAuth() throws IOException, InterruptedException {
HttpClient client = HttpClient.newBuilder()
.connectTimeout(Duration.ofMillis(5000))
.authenticator(new Authenticator() {
@Override
protected PasswordAuthentication getPasswordAuthentication() {
return new PasswordAuthentication("admin","password".toCharArray());
}
})
.build();

HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create("http://localhost:8080/json/info"))
.timeout(Duration.ofMillis(5009))
.build();

HttpResponse<String> response =
client.send(request, HttpResponse.BodyHandlers.ofString());

System.out.println(response.statusCode());
System.out.println(response.body());
}
複製程式碼

authenticator可以用來設定HTTP authentication，比如Basic authentication
雖然Basic authentication也可以自己設定header，不過通過authenticator省得自己去構造header

設定header

@Test
public void testCookies() throws IOException, InterruptedException {
HttpClient client = HttpClient.newBuilder()
.connectTimeout(Duration.ofMillis(5000))
.build();
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create("http://localhost:8080/json/cookie"))
.header("Cookie","JSESSIONID=4f994730-32d7-4e22-a18b-25667ddeb636; userId=java11")
.timeout(Duration.ofMillis(5009))
.build();
HttpResponse<String> response =
client.send(request, HttpResponse.BodyHandlers.ofString());

System.out.println(response.statusCode());
System.out.println(response.body());
}
複製程式碼

通過request可以自己設定header

GET

同步

@Test
public void testSyncGet() throws IOException, InterruptedException {
HttpClient client = HttpClient.newHttpClient();
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create("https://www.baidu.com"))
.build();

HttpResponse<String> response =
client.send(request, HttpResponse.BodyHandlers.ofString());

System.out.println(response.body());
}
複製程式碼

非同步

@Test
public void testAsyncGet() throws ExecutionException, InterruptedException {
HttpClient client = HttpClient.newHttpClient();
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create("https://www.baidu.com"))
.build();

CompletableFuture<String> result = client.sendAsync(request, HttpResponse.BodyHandlers.ofString())
.thenApply(HttpResponse::body);
System.out.println(result.get());
}
複製程式碼

POST表單

@Test
public void testPostForm() throws IOException, InterruptedException {
HttpClient client = HttpClient.newBuilder().build();
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create("http://www.w3school.com.cn/demo/demo_form.asp"))
.header("Content-Type","application/x-www-form-urlencoded")
.POST(HttpRequest.BodyPublishers.ofString("name1=value1&name2=value2"))
.build();

HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString());
System.out.println(response.statusCode());
}
複製程式碼

header指定內容是表單型別，然後通過BodyPublishers.ofString傳遞表單資料，需要自己構建表單引數

POST JSON

@Test
public void testPostJsonGetJson() throws ExecutionException, InterruptedException, JsonProcessingException {
ObjectMapper objectMapper = new ObjectMapper();
StockDto dto = new StockDto();
dto.setName("hj");
dto.setSymbol("hj");
dto.setType(StockDto.StockType.SH);
String requestBody = objectMapper
.writerWithDefaultPrettyPrinter()
.writeValueAsString(dto);

HttpRequest request = HttpRequest.newBuilder(URI.create("http://localhost:8080/json/demo"))
.header("Content-Type", "application/json")
.POST(HttpRequest.BodyPublishers.ofString(requestBody))
.build();

CompletableFuture<StockDto> result = HttpClient.newHttpClient()
.sendAsync(request, HttpResponse.BodyHandlers.ofString())
.thenApply(HttpResponse::body)
.thenApply(body -> {
try {
return objectMapper.readValue(body,StockDto.class);
} catch (IOException e) {
return new StockDto();
}
});
System.out.println(result.get());
}
複製程式碼

post json的話，body自己json化為string，然後header指定是json格式

檔案上傳

@Test
public void testUploadFile() throws IOException, InterruptedException, URISyntaxException {
HttpClient client = HttpClient.newHttpClient();
Path path = Path.of(getClass().getClassLoader().getResource("body.txt").toURI());
File file = path.toFile();

String multipartFormDataBoundary = "Java11HttpClientFormBoundary";
org.apache.http.HttpEntity multipartEntity = MultipartEntityBuilder.create()
.addPart("file", new FileBody(file, ContentType.DEFAULT_BINARY))
.setBoundary(multipartFormDataBoundary) //要設定，否則阻塞
.build();

HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create("http://localhost:8080/file/upload"))
.header("Content-Type", "multipart/form-data; boundary=" + multipartFormDataBoundary)
.POST(HttpRequest.BodyPublishers.ofInputStream(() -> {
try {
return multipartEntity.getContent();
} catch (IOException e) {
e.printStackTrace();
throw new RuntimeException(e);
}
}))
.build();

HttpResponse<String> response =
client.send(request, HttpResponse.BodyHandlers.ofString());

System.out.println(response.body());
}
複製程式碼

httpclient及httpmime

檔案下載

@Test
public void testAsyncDownload() throws ExecutionException, InterruptedException {
HttpClient client = HttpClient.newHttpClient();
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create("http://localhost:8080/file/download"))
.build();

CompletableFuture<Path> result = client.sendAsync(request, HttpResponse.BodyHandlers.ofFile(Paths.get("/tmp/body.txt")))
.thenApply(HttpResponse::body);
System.out.println(result.get());
}
複製程式碼

使用HttpResponse.BodyHandlers.ofFile來接收檔案

併發請求

@Test
public void testConcurrentRequests(){
HttpClient client = HttpClient.newHttpClient();
List<String> urls = List.of("http://www.baidu.com","http://www.alibaba.com/","http://www.tencent.com");
List<HttpRequest> requests = urls.stream()
.map(url -> HttpRequest.newBuilder(URI.create(url)))
.map(reqBuilder -> reqBuilder.build())
.collect(Collectors.toList());

List<CompletableFuture<HttpResponse<String>>> futures = requests.stream()
.map(request -> client.sendAsync(request, HttpResponse.BodyHandlers.ofString()))
.collect(Collectors.toList());
futures.stream()
.forEach(e -> e.whenComplete((resp,err) -> {
if(err != null){
err.printStackTrace();
}else{
System.out.println(resp.body());
System.out.println(resp.statusCode());
}
}));
CompletableFuture.allOf(futures
.toArray(CompletableFuture<?>[]::new))
.join();
}
複製程式碼

sendAsync方法返回的是CompletableFuture，可以方便地進行轉換、組合等操作
這裡使用CompletableFuture.allOf組合在一起，最後呼叫join等待所有future完成

錯誤處理

@Test
public void testHandleException() throws ExecutionException, InterruptedException {
HttpClient client = HttpClient.newBuilder()
.connectTimeout(Duration.ofMillis(5000))
.build();
HttpRequest request = HttpRequest.newBuilder()
.uri(URI.create("https://twitter.com"))
.build();

CompletableFuture<String> result = client.sendAsync(request, HttpResponse.BodyHandlers.ofString())
//.whenComplete((resp,err) -> {
//if(err != null){
//err.printStackTrace();
//}else{
//System.out.println(resp.body());
//System.out.println(resp.statusCode());
//}
//})
.thenApply(HttpResponse::body)
.exceptionally(err -> {
err.printStackTrace();
return "fallback";
});
System.out.println(result.get());
}
複製程式碼

HttpClient非同步請求返回的是CompletableFuture<HttpResponse>，其自帶exceptionally方法可以用來做fallback處理
另外值得注意的是HttpClient不像WebClient那樣，它沒有對4xx或5xx的狀態碼丟擲異常，需要自己根據情況來處理，手動檢測狀態碼丟擲異常或者返回其他內容

HTTP2

@Test
public void testHttp2() throws URISyntaxException {
HttpClient.newBuilder()
.followRedirects(HttpClient.Redirect.NEVER)
.version(HttpClient.Version.HTTP_2)
.build()
.sendAsync(HttpRequest.newBuilder()
.uri(new URI("https://http2.akamai.com/demo"))
.GET()
.build(),
HttpResponse.BodyHandlers.ofString())
.whenComplete((resp,t) -> {
if(t != null){
t.printStackTrace();
}else{
System.out.println(resp.version());
System.out.println(resp.statusCode());
}
}).join();
}
複製程式碼

執行之後可以看到返回的response的version為HTTP_2

Socket/">WebSocket

@Test
public void testWebSocket() throws InterruptedException {
HttpClient client = HttpClient.newHttpClient();
WebSocket webSocket = client.newWebSocketBuilder()
.buildAsync(URI.create("ws://localhost:8080/echo"), new WebSocket.Listener() {

@Override
public CompletionStage<?> onText(WebSocket webSocket, CharSequence data, boolean last) {
// request one more
webSocket.request(1);

// Print the message when it's available
return CompletableFuture.completedFuture(data)
.thenAccept(System.out::println);
}
}).join();
webSocket.sendText("hello ", false);
webSocket.sendText("world ",true);

TimeUnit.SECONDS.sleep(10);
webSocket.sendClose(WebSocket.NORMAL_CLOSURE, "ok").join();
}
複製程式碼

HttpClient支援HTTP2，也包含了WebSocket，通過newWebSocketBuilder去構造WebSocket
傳入listener進行接收訊息，要發訊息的話，使用WebSocket來發送，關閉使用sendClose方法

reactive streams

HttpClient本身就是reactive的，支援reactive streams，這裡舉ResponseSubscribers.ByteArraySubscriber的原始碼看看： java.net.http/jdk/internal/net/http/ResponseSubscribers.java

public static class ByteArraySubscriber<T> implements BodySubscriber<T> {
private final Function<byte[], T> finisher;
private final CompletableFuture<T> result = new MinimalFuture<>();
private final List<ByteBuffer> received = new ArrayList<>();

private volatile Flow.Subscription subscription;

public ByteArraySubscriber(Function<byte[],T> finisher) {
this.finisher = finisher;
}

@Override
public void onSubscribe(Flow.Subscription subscription) {
if (this.subscription != null) {
subscription.cancel();
return;
}
this.subscription = subscription;
// We can handle whatever you've got
subscription.request(Long.MAX_VALUE);
}

@Override
public void onNext(List<ByteBuffer> items) {
// incoming buffers are allocated by http client internally,
// and won't be used anywhere except this place.
// So it's free simply to store them for further processing.
assert Utils.hasRemaining(items);
received.addAll(items);
}

@Override
public void onError(Throwable throwable) {
received.clear();
result.completeExceptionally(throwable);
}

static private byte[] join(List<ByteBuffer> bytes) {
int size = Utils.remaining(bytes, Integer.MAX_VALUE);
byte[] res = new byte[size];
int from = 0;
for (ByteBuffer b : bytes) {
int l = b.remaining();
b.get(res, from, l);
from += l;
}
return res;
}

@Override
public void onComplete() {
try {
result.complete(finisher.apply(join(received)));
received.clear();
} catch (IllegalArgumentException e) {
result.completeExceptionally(e);
}
}

@Override
public CompletionStage<T> getBody() {
return result;
}
}
複製程式碼

BodySubscriber介面繼承了Flow.Subscriber<List>介面
這裡的Subscription來自Flow類，該類是java9引入的，裡頭包含了支援Reactive Streams的實現

小結

HttpClient在Java11從incubator變為正式版，相對於傳統的HttpUrlConnection其提升可不是一點半點，不僅支援非同步，也支援reactive streams，同時也支援了HTTP2以及WebSocket，非常值得大家使用。