Java11新特性之HttpClient小试牛刀
作者:codecraft
序
本文主要研究一下Java11的HttpClient的基本使用。
变化
- 从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并没有提供类似WebClient那种现成的BodyInserters.fromMultipartData方法,因此这里需要自己转换
- 这里使用org.apache.httpcomponents(httpclient及httpmime)的MultipartEntityBuilder构建multipartEntity,最后通过HttpRequest.BodyPublishers.ofInputStream来传递内容
- 这里header要指定Content-Type值为multipart/form-data以及boundary的值,否则服务端可能无法解析
文件下载
@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<T>>,其自带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
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<ByteBuffer>>接口
- 这里的Subscription来自Flow类,该类是java9引入的,里头包含了支持Reactive Streams的实现
小结
HttpClient在Java11从incubator变为正式版,相对于传统的HttpUrlConnection其提升可不是一点半点,不仅支持异步,也支持reactive streams,同时也支持了HTTP2以及WebSocket,非常值得大家使用。
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持脚本之家。