聊聊jdk httpclient的connect timeout异常-白红宇

聊聊jdk httpclient的connect timeout异常

阅读量：6690 次

发布时间：2019-06-25

本文共 21727 字，大约阅读时间需要 72 分钟。

序

本文主要研究一下httpclient的connect timeout异常

实例代码

@Test    public void testConnectTimeout() throws IOException, InterruptedException {        HttpClient client = HttpClient.newBuilder()                .build();        HttpRequest request = HttpRequest.newBuilder()                .uri(URI.create("https://twitter.com"))                .build();        long start = System.currentTimeMillis();        try{            HttpResponse
    
      result = client.send(request, HttpResponse.BodyHandlers.ofString());            System.out.println(result.body());        }finally {            long cost = System.currentTimeMillis() - start;            System.out.println("cost:"+cost);        }    }

异常日志如下：

cost:75814java.net.ConnectException: Operation timed out	at java.net.http/jdk.internal.net.http.HttpClientImpl.send(HttpClientImpl.java:561)	at java.net.http/jdk.internal.net.http.HttpClientFacade.send(HttpClientFacade.java:119)	at com.example.HttpClientTest.testConnectTimeout(HttpClientTest.java:464)	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)	at java.base/java.lang.reflect.Method.invoke(Method.java:566)	at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)	at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)	at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)	at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)	at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:325)	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:78)	at org.junit.runners.BlockJUnit4ClassRunner.runChild(BlockJUnit4ClassRunner.java:57)	at org.junit.runners.ParentRunner$3.run(ParentRunner.java:290)	at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:71)	at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:288)	at org.junit.runners.ParentRunner.access$000(ParentRunner.java:58)	at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:268)	at org.junit.runners.ParentRunner.run(ParentRunner.java:363)	at org.junit.runner.JUnitCore.run(JUnitCore.java:137)	at com.intellij.junit4.JUnit4IdeaTestRunner.startRunnerWithArgs(JUnit4IdeaTestRunner.java:68)	at com.intellij.rt.execution.junit.IdeaTestRunner$Repeater.startRunnerWithArgs(IdeaTestRunner.java:47)	at com.intellij.rt.execution.junit.JUnitStarter.prepareStreamsAndStart(JUnitStarter.java:242)	at com.intellij.rt.execution.junit.JUnitStarter.main(JUnitStarter.java:70)	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)	at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)	at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)	at java.base/java.lang.reflect.Method.invoke(Method.java:566)	at com.intellij.rt.execution.application.AppMainV2.main(AppMainV2.java:131)Caused by: java.net.ConnectException: Operation timed out	at java.base/sun.nio.ch.SocketChannelImpl.checkConnect(Native Method)	at java.base/sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:779)	at java.net.http/jdk.internal.net.http.PlainHttpConnection$ConnectEvent.handle(PlainHttpConnection.java:128)	at java.net.http/jdk.internal.net.http.HttpClientImpl$SelectorManager.handleEvent(HttpClientImpl.java:957)	at java.net.http/jdk.internal.net.http.HttpClientImpl$SelectorManager.lambda$run$3(HttpClientImpl.java:912)	at java.base/java.util.ArrayList.forEach(ArrayList.java:1540)	at java.net.http/jdk.internal.net.http.HttpClientImpl$SelectorManager.run(HttpClientImpl.java:912)

Exchange.responseAsync

java.net.http/jdk/internal/net/http/Exchange.java

public CompletableFuture
    
      responseAsync() {        return responseAsyncImpl(null);    }    CompletableFuture
     
       responseAsyncImpl(HttpConnection connection) {        SecurityException e = checkPermissions();        if (e != null) {            return MinimalFuture.failedFuture(e);        } else {            return responseAsyncImpl0(connection);        }    }    CompletableFuture
      
        responseAsyncImpl0(HttpConnection connection) {        Function
       
        
         , CompletableFuture
         
          > after407Check; bodyIgnored = null; if (request.expectContinue()) { request.addSystemHeader("Expect", "100-Continue"); Log.logTrace("Sending Expect: 100-Continue"); // wait for 100-Continue before sending body after407Check = this::expectContinue; } else { // send request body and proceed. after407Check = this::sendRequestBody; } // The ProxyAuthorizationRequired can be triggered either by // establishExchange (case of HTTP/2 SSL tunneling through HTTP/1.1 proxy // or by sendHeaderAsync (case of HTTP/1.1 SSL tunneling through HTTP/1.1 proxy // Therefore we handle it with a call to this checkFor407(...) after these // two places. Function
          
           
            , CompletableFuture
            
             > afterExch407Check = (ex) -> ex.sendHeadersAsync() .handle((r,t) -> this.checkFor407(r, t, after407Check)) .thenCompose(Function.identity()); return establishExchange(connection) .handle((r,t) -> this.checkFor407(r,t, afterExch407Check)) .thenCompose(Function.identity()); } // get/set the exchange impl, solving race condition issues with // potential concurrent calls to cancel() or cancel(IOException) private CompletableFuture
             > establishExchange(HttpConnection connection) { if (debug.on()) { debug.log("establishing exchange for %s,%n\t proxy=%s", request, request.proxy()); } // check if we have been cancelled first. Throwable t = getCancelCause(); checkCancelled(); if (t != null) { return MinimalFuture.failedFuture(t); } CompletableFuture
             > cf, res; cf = ExchangeImpl.get(this, connection); // We should probably use a VarHandle to get/set exchangeCF // instead - as we need CAS semantics. synchronized (this) { exchangeCF = cf; }; res = cf.whenComplete((r,x) -> { synchronized(Exchange.this) { if (exchangeCF == cf) exchangeCF = null; } }); checkCancelled(); return res.thenCompose((eimpl) -> { // recheck for cancelled, in case of race conditions exchImpl = eimpl; IOException tt = getCancelCause(); checkCancelled(); if (tt != null) { return MinimalFuture.failedFuture(tt); } else { // Now we're good to go. Because exchImpl is no longer // null cancel() will be able to propagate directly to // the impl after this point ( if needed ). return MinimalFuture.completedFuture(eimpl); } }); }

responseAsync最后调用ExchangeImpl.get(this, connection)

ExchangeImpl.get

java.net.http/jdk/internal/net/http/ExchangeImpl.java

/**     * Initiates a new exchange and assigns it to a connection if one exists     * already. connection usually null.     */    static  CompletableFuture
     >    get(Exchange exchange, HttpConnection connection)    {        if (exchange.version() == HTTP_1_1) {            if (debug.on())                debug.log("get: HTTP/1.1: new Http1Exchange");            return createHttp1Exchange(exchange, connection);        } else {            Http2ClientImpl c2 = exchange.client().client2(); // #### improve            HttpRequestImpl request = exchange.request();            CompletableFuture
      
        c2f = c2.getConnectionFor(request, exchange);            if (debug.on())                debug.log("get: Trying to get HTTP/2 connection");            return c2f.handle((h2c, t) -> createExchangeImpl(h2c, t, exchange, connection))                    .thenCompose(Function.identity());        }    }

这里调用Http2ClientImpl.getConnectionFor获取连接

Http2ClientImpl.getConnectionFor

java.net.http/jdk/internal/net/http/Http2ClientImpl.java

/**     * When HTTP/2 requested only. The following describes the aggregate behavior including the     * calling code. In all cases, the HTTP2 connection cache     * is checked first for a suitable connection and that is returned if available.     * If not, a new connection is opened, except in https case when a previous negotiate failed.     * In that case, we want to continue using http/1.1. When a connection is to be opened and     * if multiple requests are sent in parallel then each will open a new connection.     *     * If negotiation/upgrade succeeds then     * one connection will be put in the cache and the others will be closed     * after the initial request completes (not strictly necessary for h2, only for h2c)     *     * If negotiate/upgrade fails, then any opened connections remain open (as http/1.1)     * and will be used and cached in the http/1 cache. Note, this method handles the     * https failure case only (by completing the CF with an ALPN exception, handled externally)     * The h2c upgrade is handled externally also.     *     * Specific CF behavior of this method.     * 1. completes with ALPN exception: h2 negotiate failed for first time. failure recorded.     * 2. completes with other exception: failure not recorded. Caller must handle     * 3. completes normally with null: no connection in cache for h2c or h2 failed previously     * 4. completes normally with connection: h2 or h2c connection in cache. Use it.     */    CompletableFuture
    
      getConnectionFor(HttpRequestImpl req,                                                        Exchange
      exchange) {        URI uri = req.uri();        InetSocketAddress proxy = req.proxy();        String key = Http2Connection.keyFor(uri, proxy);        synchronized (this) {            Http2Connection connection = connections.get(key);            if (connection != null) {                try {                    if (connection.closed || !connection.reserveStream(true)) {                        if (debug.on())                            debug.log("removing found closed or closing connection: %s", connection);                        deleteConnection(connection);                    } else {                        // fast path if connection already exists                        if (debug.on())                            debug.log("found connection in the pool: %s", connection);                        return MinimalFuture.completedFuture(connection);                    }                } catch (IOException e) {                    // thrown by connection.reserveStream()                    return MinimalFuture.failedFuture(e);                }            }            if (!req.secure() || failures.contains(key)) {                // secure: negotiate failed before. Use http/1.1                // !secure: no connection available in cache. Attempt upgrade                if (debug.on()) debug.log("not found in connection pool");                return MinimalFuture.completedFuture(null);            }        }        return Http2Connection                .createAsync(req, this, exchange)                .whenComplete((conn, t) -> {                    synchronized (Http2ClientImpl.this) {                        if (conn != null) {                            try {                                conn.reserveStream(true);                            } catch (IOException e) {                                throw new UncheckedIOException(e); // shouldn't happen                            }                            offerConnection(conn);                        } else {                            Throwable cause = Utils.getCompletionCause(t);                            if (cause instanceof Http2Connection.ALPNException)                                failures.add(key);                        }                    }                });    }

如果没有连接会新创建一个，走的是Http2Connection.createAsync

Http2Connection.createAsync

java.net.http/jdk/internal/net/http/Http2Connection.java

// Requires TLS handshake. So, is really async    static CompletableFuture
    
      createAsync(HttpRequestImpl request,                                                          Http2ClientImpl h2client,                                                          Exchange
      exchange) {        assert request.secure();        AbstractAsyncSSLConnection connection = (AbstractAsyncSSLConnection)        HttpConnection.getConnection(request.getAddress(),                                     h2client.client(),                                     request,                                     HttpClient.Version.HTTP_2);        // Expose the underlying connection to the exchange's aborter so it can        // be closed if a timeout occurs.        exchange.connectionAborter.connection(connection);        return connection.connectAsync(exchange)                  .thenCompose(unused -> connection.finishConnect())                  .thenCompose(unused -> checkSSLConfig(connection))                  .thenCompose(notused-> {                      CompletableFuture
     
       cf = new MinimalFuture<>();                      try {                          Http2Connection hc = new Http2Connection(request, h2client, connection);                          cf.complete(hc);                      } catch (IOException e) {                          cf.completeExceptionally(e);                      }                      return cf; } );    }

这里先是调用了HttpConnection.getConnection获取连接，然后调用connectAsync进行连接

AsyncSSLConnection

java.net.http/jdk/internal/net/http/AsyncSSLConnection.java

@Override    public CompletableFuture
    
      connectAsync(Exchange
      exchange) {        return plainConnection                .connectAsync(exchange)                .thenApply( unused -> {                    // create the SSLTube wrapping the SocketTube, with the given engine                    flow = new SSLTube(engine,                                       client().theExecutor(),                                       client().getSSLBufferSupplier()::recycle,                                       plainConnection.getConnectionFlow());                    return null; } );    }

这里委托给plainConnection.connectAsync

PlainHttpConnection.connectAsync

java.net.http/jdk/internal/net/http/PlainHttpConnection.java

@Override    public CompletableFuture
    
      connectAsync(Exchange
      exchange) {        CompletableFuture
     
       cf = new MinimalFuture<>();        try {            assert !connected : "Already connected";            assert !chan.isBlocking() : "Unexpected blocking channel";            boolean finished;            connectTimerEvent = newConnectTimer(exchange, cf);            if (connectTimerEvent != null) {                if (debug.on())                    debug.log("registering connect timer: " + connectTimerEvent);                client().registerTimer(connectTimerEvent);            }            PrivilegedExceptionAction
      
        pa =                    () -> chan.connect(Utils.resolveAddress(address));            try {                 finished = AccessController.doPrivileged(pa);            } catch (PrivilegedActionException e) {               throw e.getCause();            }            if (finished) {                if (debug.on()) debug.log("connect finished without blocking");                cf.complete(null);            } else {                if (debug.on()) debug.log("registering connect event");                client().registerEvent(new ConnectEvent(cf));            }        } catch (Throwable throwable) {            cf.completeExceptionally(Utils.toConnectException(throwable));            try {                close();            } catch (Exception x) {                if (debug.on())                    debug.log("Failed to close channel after unsuccessful connect");            }        }        return cf;    }

这里如果client有设置connectTimeout的话，则会创建一个connectTimerEvent

调用chan.connect进行连接，如果连接未完成，则注册ConnectEvent

SocketChannelImpl.connect

java.base/sun/nio/ch/SocketChannelImpl.java

@Override    public boolean connect(SocketAddress sa) throws IOException {        InetSocketAddress isa = Net.checkAddress(sa);        SecurityManager sm = System.getSecurityManager();        if (sm != null)            sm.checkConnect(isa.getAddress().getHostAddress(), isa.getPort());        InetAddress ia = isa.getAddress();        if (ia.isAnyLocalAddress())            ia = InetAddress.getLocalHost();        try {            readLock.lock();            try {                writeLock.lock();                try {                    int n = 0;                    boolean blocking = isBlocking();                    try {                        beginConnect(blocking, isa);                        do {                            n = Net.connect(fd, ia, isa.getPort());                        } while (n == IOStatus.INTERRUPTED && isOpen());                    } finally {                        endConnect(blocking, (n > 0));                    }                    assert IOStatus.check(n);                    return n > 0;                } finally {                    writeLock.unlock();                }            } finally {                readLock.unlock();            }        } catch (IOException ioe) {            // connect failed, close the channel            close();            throw SocketExceptions.of(ioe, isa);        }    }

通过Net.connect调用本地方法进行连接

ConnectEvent

java.net.http/jdk/internal/net/http/PlainHttpConnection.java

final class ConnectEvent extends AsyncEvent {        private final CompletableFuture
    
      cf;        ConnectEvent(CompletableFuture
     
       cf) {            this.cf = cf;        }        @Override        public SelectableChannel channel() {            return chan;        }        @Override        public int interestOps() {            return SelectionKey.OP_CONNECT;        }        @Override        public void handle() {            try {                assert !connected : "Already connected";                assert !chan.isBlocking() : "Unexpected blocking channel";                if (debug.on())                    debug.log("ConnectEvent: finishing connect");                boolean finished = chan.finishConnect();                assert finished : "Expected channel to be connected";                if (debug.on())                    debug.log("ConnectEvent: connect finished: %s Local addr: %s",                              finished, chan.getLocalAddress());                // complete async since the event runs on the SelectorManager thread                cf.completeAsync(() -> null, client().theExecutor());            } catch (Throwable e) {                Throwable t = Utils.toConnectException(e);                client().theExecutor().execute( () -> cf.completeExceptionally(t));                close();            }        }        @Override        public void abort(IOException ioe) {            client().theExecutor().execute( () -> cf.completeExceptionally(ioe));            close();        }    }

SelectorManager对准备好的事件触发handle操作，对于ConnectEvent，就是调用ConnectEvent.handle

ConnectEvent的handle方法执行chan.finishConnect()，如果捕获到异常，则调用cf.completeExceptionally(t)

SocketChannelImpl.finishConnect

java.base/sun/nio/ch/SocketChannelImpl.java

@Override    public boolean finishConnect() throws IOException {        try {            readLock.lock();            try {                writeLock.lock();                try {                    // no-op if already connected                    if (isConnected())                        return true;                    boolean blocking = isBlocking();                    boolean connected = false;                    try {                        beginFinishConnect(blocking);                        int n = 0;                        if (blocking) {                            do {                                n = checkConnect(fd, true);                            } while ((n == 0 || n == IOStatus.INTERRUPTED) && isOpen());                        } else {                            n = checkConnect(fd, false);                        }                        connected = (n > 0);                    } finally {                        endFinishConnect(blocking, connected);                    }                    assert (blocking && connected) ^ !blocking;                    return connected;                } finally {                    writeLock.unlock();                }            } finally {                readLock.unlock();            }        } catch (IOException ioe) {            // connect failed, close the channel            close();            throw SocketExceptions.of(ioe, remoteAddress);        }    }

checkConnect是一个本地方法，如果是连接超时，则抛出java.net.ConnectException: Operation timed out

tcp连接syn超时(`net.ipv4.tcp_syn_retries`)

当client端与server端建立连接，client发出syn包，如果等待一定时间没有收到server端发来的SYN+ACK，则会进行重试，重试次数由具体由net.ipv4.tcp_syn_retries决定

/ # sysctl -a | grep tcp_syn_retriessysctl: error reading key 'net.ipv6.conf.all.stable_secret': I/O errornet.ipv4.tcp_syn_retries = 6sysctl: error reading key 'net.ipv6.conf.default.stable_secret': I/O errorsysctl: error reading key 'net.ipv6.conf.eth0.stable_secret': I/O errorsysctl: error reading key 'net.ipv6.conf.lo.stable_secret': I/O error

linux默认是6次，第一次发送等待2^0秒没收到回包则重试第一次，之后等待2^1，以此类推，第六次重试等待2^6秒，因此一共是1s+2s+4s+8s+16s+32s+64s=127s，因而在linux平台下，如果httpclient没有设置connect timeout，则依赖系统tcp的syn超时，即127s之后超时，java的本地调用抛出java.net.ConnectException: Operation timed out