java - Java 11 HttpClient取消http请求

Question

我正在尝试通过新的 Java 11 HttpClient 取消 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); 行后立即取消。因此，控制台中最后打印的行应该是 isCancelled: true

但是，当我运行这段代码时，我看到了这样的东西:

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

This means, that request task still running after I cancel it...So, that is the right way to cancel request?

UPD

Right way to cancel request is (As daniel suggested, + UPD2: avoiding NPE on cancel() method invoke):

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

您可以使用 java.net.http.HttpClient API 取消 HTTP 请求，方法是取消传递给响应的 BodySubscriber 的 Flow.Subscription 对象。简单地包装提供的 BodyHandler/BodySubscriber 实现之一以获取订阅对象应该相对容易。不幸的是，客户端返回的 CompletableFuture 的 cancel 方法与 Flow.Subscription 的 cancel 方法之间没有任何关系 传递给 BodySubscriber。取消请求的正确方法是通过订阅的 cancel 方法。

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

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