从JDK11新增HttpClient谈谈非阻塞模型

2022-06-02 23:54:49

北京时间 9 月 26 日，Oracle 官方宣布 Java 11 正式发布

一、JDK HTTP Client介绍

JDK11中的17个新特性

JDK11中引入HTTP Client的动机

既有的HttpURLConnection存在许多问题

其基类URLConnection当初是设计为支持多协议，但其中大多已经成为非主流（ftp, gopher…）
API的设计早于HTTP/1.1，过度抽象
难以使用，存在许多没有文档化的行为
它只支持阻塞模式（每个请求/响应占用一个线程）

HTTP Client发展史

在JDK11 HTTP Client出现之前

在此之前，可以使用以下工具作为Http客户端

JDK HttpURLConnection
Apache HttpClient
Okhttp
Spring Rest Template
Spring Cloud Feign
将Jetty用作客户端
使用Netty库。还

初探JDK HTTP Client

我们来看一段HTTP Client的常规用法的样例 ——
执行GET请求，然后输出响应体（Response Body）。

HttpClient client = HttpClient.newHttpClient();
HttpRequest request = HttpRequest.newBuilder()
      .uri(URI.create("http://openjdk.java.net/"))
      .build();
client.sendAsync(request, asString())
      .thenApply(HttpResponse::body)
      .thenAccept(System.out::println)
      .join();

第一步：创建HttpClient

一般使用JDK 11中的HttpClient的第一步是创建HttpClient对象并进行配置。

指定协议（http/1.1或者http/2）
转发（redirect）
代理（proxy）
认证（authenticator）

HttpClient client = HttpClient.newBuilder()
      .version(Version.HTTP_2)
      .followRedirects(Redirect.SAME_PROTOCOL)
      .proxy(ProxySelector.of(new InetSocketAddress("www-proxy.com", 8080)))
      .authenticator(Authenticator.getDefault())
      .build();

第二步：创建HttpRequest

从HttpRequest的builder组建request

请求URI
请求method（GET, PUT, POST）
请求体（request body）
Timeout
请求头（request header）

HttpRequest request = HttpRequest.newBuilder()
      .uri(URI.create("http://openjdk.java.net/"))
      .timeout(Duration.ofMinutes(1))
      .header("Content-Type", "application/json")
      .POST(BodyPublisher.fromFile(Paths.get("file.json")))
      .build()

第三步：send

http client可以用来发送多个http request
请求可以被以同步或异步方式发送

1. 同步发送

同步发送API阻塞直到HttpResponse返回

HttpResponse<String> response =
      client.send(request, BodyHandler.asString());
System.out.println(response.statusCode());
System.out.println(response.body());

2. 异步发送

异步发送API立即返回一个CompletableFuture
当它完成的时候会获得一个HttpResponse

client.sendAsync(request, BodyHandler.asString())
      .thenApply(response -> { System.out.println(response.statusCode());
                               return response; } )
      .thenApply(HttpResponse::body)
      .thenAccept(System.out::println);

※CompletableFuture是在java8中加入的，支持组合式异步编程

二、从提升单机并发处理能力的技术来看HttpClient的实现细节

提升单机并发处理能力的技术

多CPU/多核
多线程
非阻塞（non-blocking）

Java NIO

Java NIO为Java带来了非阻塞模型。

Lambda表达式

Lambda表达式可以方便地利用多CPU。
Lambda表达式让代码更加具有可读性

回调

假设以下代码是一个聊天应用服务器的一部分，该应用以Vert.x框架实现。
(Eclipse Vert.x is a tool-kit for building reactive applications on the JVM.)
向connectHandler方法输入一个Lambda表达式，每当有用户连接到聊天应用时，都会调用该Lambda表达式。这就是一个回调。
这种方式的好处是，应用不必控制线程模型——Vert.x框架为我们管理线程，打理好一切相关复杂性，程序员只考虑和回调就够了。

vertx.createServer()
    .connectHandler(socket -> {
        socket.dataHandler(new User(socket, this));
    }).listen(10_000);

注意，这种设计里，不共享任何状态。对象之间通过向事件总线发送消息通信，根本不需要在代码中添加锁或使用synchronized关键字。并发编程变得更加简单。

Future

大量的回调会怎样？请看以下伪代码

(1)->{
    (2)->{
        (3)->{
            (4)->{}
        }
    }
}

大量回调会形成“末日金字塔”。

如何破解？使用Future

Future1=(1)->{}
Future2=(Future1.get())->{}
Future3=(Future2.get())->{}
Future4=(Future3.get())->{}

但这会造成原本期望的并行处理，变成了串行处理，带来了性能问题。
我们真正需要的是将Future和回调联合起来使用。下面将要讲的CompletableFuture就是结合了Future和回调，其要点是组合不同实例而无需担心末日金字塔问题。

CompletableFuture

(new CompletableFuture()).thenCompose((1)->{})
    .thenCompose((2)->{})
    .thenCompose((3)->{})
    .thenCompose((4)->{})
    .join()

Reactive Streams

Reactive Streams是一个倡议，它提倡提供一种带有非阻塞背压的异步流处理的标准（Reactive Streams is an initiative to provide a standard for asynchronous stream processing with non-blocking back pressure）。

JDK 9中的java.util.concurrent.Flow中的概念，与Reactive Streams是一对一对等的。java.util.concurrent.Flow是Reactive Streams标准的实现之一。

多CPU的并行机制让处理海量数据的速度更快，消息传递和响应式编程让有限的并行运行的线程执行更多的I/O操作。

HttpClient的实现细节

请求响应的body与reactive streams

请求响应的body暴露为reactive streams
http client是请求的body的消费者
http client是响应的body的生产者

HttpRequest内部

public abstract class HttpRequest {
    ...
    public interface BodyPublisher
                extends Flow.Publisher<ByteBuffer> { ... }
}

HttpResponse的内部

public abstract class HttpResponse<T> {
    ...
    public interface BodyHandler<T> {
        BodySubscriber<T> apply(int statusCode, HttpHeaders responseHeaders);
    }

    public interface BodySubscriber<T>
        extends Flow.Subscriber<List<ByteBuffer>> { ... }
}

码农公寓