Java 11 HttpClient中http请求的取消

Question

我正试图通过新的Java11HttpClient取消http请求。

这是我的测试代码:

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;

public class App {

    public static void main(String... args) throws InterruptedException {
        HttpClient client = HttpClient.newBuilder().build();

        URI uri = URI.create("http://releases.ubuntu.com/18.04.2/ubuntu-18.04.2-desktop-amd64.iso");
        HttpRequest request = HttpRequest.newBuilder().uri(uri).GET().build();

        var bodyHandler = HttpResponse.BodyHandlers.ofByteArrayConsumer(b -> System.out.println("#"));
        var future = client.sendAsync(request, bodyHandler);
        Thread.sleep(1000);

        future.cancel(true);
        System.out.println("\r\n----------CANCEL!!!------------");
        System.out.println("\r\nisCancelled: " + future.isCancelled());
        Thread.sleep(250);
    }
}

我预计，该请求任务将在 future.cancel（true）; 行调用后立即被取消。因此，控制台中最后打印的行应该是Cancelled：true

但是，当我运行这段代码时，我看到类似这样的内容:

####################################################################################################
----------CANCEL!!!------------
####
isCancelled: true
#######################################################################################################################################################

这意味着，在我取消请求后，请求任务仍在运行…所以，这是取消请求的正确方法吗？

UPD

取消请求的正确方法是（正如daniel所建议的，UPD2：在cancel（）方法调用中避免NPE）：

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.http.HttpResponse.BodyHandler;
import java.net.http.HttpResponse.BodySubscriber;
import java.net.http.HttpResponse.ResponseInfo;
import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Flow.Subscription;

public class App {

    private static class SubscriberWrapper implements BodySubscriber<Void> {
        private final CountDownLatch latch;
        private final BodySubscriber<Void> subscriber;
        private Subscription subscription;

        private SubscriberWrapper(BodySubscriber<Void> subscriber, CountDownLatch latch) {
            this.subscriber = subscriber;
            this.latch = latch;
        }

        @Override
        public CompletionStage<Void> getBody() {
            return subscriber.getBody();
        }

        @Override
        public void onSubscribe(Subscription subscription) {
            subscriber.onSubscribe(subscription);
            this.subscription = subscription;
            latch.countDown();
        }

        @Override
        public void onNext(List<ByteBuffer> item) {
            subscriber.onNext(item);
        }

        @Override
        public void onError(Throwable throwable) {
            subscriber.onError(throwable);
        }

        @Override
        public void onComplete() {
            subscriber.onComplete();
        }

        public void cancel() {
            subscription.cancel();
            System.out.println("\r\n----------CANCEL!!!------------");
        }
    }

    private static class BodyHandlerWrapper implements BodyHandler<Void> {
        private final CountDownLatch latch = new CountDownLatch(1);
        private final BodyHandler<Void> handler;
        private SubscriberWrapper subscriberWrapper;

        private BodyHandlerWrapper(BodyHandler<Void> handler) {
            this.handler = handler;
        }

        @Override
        public BodySubscriber<Void> apply(ResponseInfo responseInfo) {
            subscriberWrapper = new SubscriberWrapper(handler.apply(responseInfo), latch);
            return subscriberWrapper;
        }

        public void cancel() {
            CompletableFuture.runAsync(() -> {
                try {
                    latch.await();
                    subscriberWrapper.cancel();
                } catch (InterruptedException e) {}
            });
        }
    }

    public static void main(String... args) throws InterruptedException, ExecutionException {
        HttpClient client = HttpClient.newBuilder().build();

        URI uri = URI.create("http://releases.ubuntu.com/18.04.2/ubuntu-18.04.2-desktop-amd64.iso");
        HttpRequest request = HttpRequest.newBuilder().uri(uri).GET().build();

        var handler = HttpResponse.BodyHandlers.ofByteArrayConsumer(b -> System.out.print("#"));
        BodyHandlerWrapper handlerWrapper = new BodyHandlerWrapper(handler);

        client.sendAsync(request, handlerWrapper).thenAccept(b -> System.out.println(b.statusCode()));
        Thread.sleep(1000);
        handlerWrapper.cancel();

        System.out.println("\r\n------Invoke cancel...---------");
        Thread.sleep(2500);
    }
}

Answer 1

至少对于同步请求，您可以中断调用httpClient.send（...）的线程

http客户端然后尽可能快地中止请求，并抛出一个中断异常。

Answer 2

同步 VS 异步

请求可以同步或异步发送。同步API会一直阻塞，直到Http响应可用

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

异步API会立即返回一个CompletableFuture，当HttpResponse可用时，它将完成。CompletableFuture是在Java 8中添加的，支持可组合异步编程。

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

未来对象

未来代表异步计算的结果。Java文档

这意味着它不是一个同步函数，并且您的假设“我预计，该请求任务将立即被取消”仅适用于同步方法。

检查未来对象的取消

如果您想检查任务是否被取消，有一个有用的< code>isCancelled()方法。

if(future.isCancelled()) {
  // Future object is cancelled, do smth
} else {
  // Future object is still running, do smth
}

sendAsync（）返回一个CompletableFuture对象

方法＜code＞sendAsync（）CompletableFuture实现了Future的接口。

您可以这样做:

client.sendAsync(request, BodyHandlers.ofString())
          .thenAccept(response -> {
       // do action when completed;
});

在技术术语中，thenAccept方法添加了一个消费者，以便在响应可用时调用。

为什么取消一种在未来将不起作用的、竞争力过强的方法

由于（与 FutureTask 不同）此类无法直接控制导致其完成的计算，因此取消被视为另一种形式的特殊完成。方法取消与 completeException（new CancelException（）） 具有相同的效果。方法 isCompletedExceptionly（） 可用于确定 CompletableFuture 是否以任何特殊方式完成。

在使用CompletionException异常完成的情况下，方法get（）和get（long， TimeUnit）抛出ExecutionException，其原因与相应的CompletionException中持有的相同。为了简化大多数上下文中的使用，此类还定义了方法connect（）和getNow（T），在这些情况下直接抛出CompletionException。

换句话说

cancel（） 方法不使用中断来执行取消，这就是它不起作用的原因。你应该使用 completeException（new CancelException（））

参考

• https://openjdk.java.net/groups/net/httpclient/intro.html
• https://docs.oracle.com/javase/8/docs/api/index.html?java/util/concurrent/Future.html

Answer 3

您可以使用java.net.HTTP取消HTTP请求。通过取消流，HttpClientAPI。Subscription对象，该对象传递给响应的＜code＞BodySubscriber。为了获得订阅对象，将提供的＜code＞BodyHandler＜/code＞或＜code＞BodySubscriber＜/code＞实现中的一个打包起来应该相对容易。不幸的是，客户端返回的CompletableFuture的cancel方法与流的。订阅传递给BodySubscriber。取消请求的正确方法是通过订阅的取消方法。

取消订阅将适用于同步（HttpClient::send）和异步（HttpClient::sendAsync）方法。但是，根据请求是通过HTTP/1.1还是HTTP/2.0发送，它会产生不同的效果（使用HTTP/1.1会导致连接关闭，使用HTTP/2.0会导致流重置）。当然，如果响应的最后一个字节已经发送到BodySubscriber，它可能根本没有效果。

更新:从Java 16开始，可以通过中断调用HttpClient::send的线程或调用HttpClient::sendAsync返回的< code>CompletableFuture上的< code>cancel(true)来取消请求。这已由JDK-8245462实施