聊聊jdk httpclient的connect timeout异常

这里调用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

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进行连接

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

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

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调用本地方法进行连接

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)

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

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

/ # sysctl -a | grep tcp_syn_retries
sysctl: error reading key "net.ipv6.conf.all.stable_secret": I/O error
net.ipv4.tcp_syn_retries = 6
sysctl: error reading key "net.ipv6.conf.default.stable_secret": I/O error
sysctl: error reading key "net.ipv6.conf.eth0.stable_secret": I/O error
sysctl: 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

如果是mac系统，根据Overriding the default Linux kernel 20-second TCP socket connect timeout的描述，超时是75s，与本实例代码输出的75814ms近似一致。

使用jdk httpclient进行连接，如果没有设置client的connectTimeout，则具体的超时时间依赖系统的tcp相关设置

如果client端sync发送超时，则依赖tcp_syn_retries的配置来决定本地方法抛出java.net.ConnectException: Operation timed out异常的时间

linux下默认tcp_syn_retries默认为6，即重试6次，一共需要1s+2s+4s+8s+16s+32s+64s=127s，若再没有收到server端发来的SYN+ACK则抛出java.net.ConnectException: Operation timed out异常

TCP协议的那些超时

Linux 建立 TCP 连接的超时时间分析

Overriding the default Linux kernel 20-second TCP socket connect timeout

设置linux中tcp默认的20秒connect超时时间

SYN丢包的几个例子