okhttp之旅(四)--RetryAndFollowUpInterceptor重定向拦截器

连接的创建是在StreamAllocation对象统筹下完成的,我们前面也说过它早在RetryAndFollowUpInterceptor就被创建了,StreamAllocation对象
主要用来管理两个关键角色:

  • RealConnection:真正建立连接的对象,利用Socket建立连接
  • ConnectionPool:连接池,用来管理和复用连接。

1 创建连接

  • 我们在前面的ConnectInterceptor分析中已经说过,connectInterceptor用来完成连接。
  • 真正的连接在RealConnection中实现,连接由连接池ConnectPool来管理,连接池最多保持5个地址的连接keep-alive,每个keep-alive时长为5分钟,并有异步线程清理无效的连接。
///主要由以下两个方法完成:
HttpCodec httpCodec = streamAllocation.newStream(client, doExtensiveHealthChecks);
RealConnection connection = streamAllocation.connection();

1.0 newStream()

  • StreamAllocation.newStream() 主要做的事情正是创建HttpCodec。
  • StreamAllocation.newStream() 根据 OkHttpClient中的设置,连接超时、读超时、写超时及连接失败是否重试,调用 findHealthyConnection() 完成 连接,即RealConnection 的创建。
  • 然后根据HTTP协议的版本创建Http1Codec或Http2Codec。
  • HttpCodec编码HTTP请求并解码HTTP响应。
  • HttpCodec提供了这样的一些操作:
  • 为发送请求而提供的,写入请求头部。
  • 为发送请求而提供的,创建请求体,以用于发送请求体数据。
  • 为发送请求而提供的,结束请求发送。
  • 为获得响应而提供的,读取响应头部。
  • 为获得响应而提供的,打开请求体,以用于后续获取请求体数据。
  • 取消请求执行。
    public HttpCodec newStream(
            OkHttpClient client, Interceptor.Chain chain, boolean doExtensiveHealthChecks) {
        int connectTimeout = chain.connectTimeoutMillis();
        int readTimeout = chain.readTimeoutMillis();
        int writeTimeout = chain.writeTimeoutMillis();
        int pingIntervalMillis = client.pingIntervalMillis();
        boolean connectionRetryEnabled = client.retryOnConnectionFailure();

        try {
            //完成 连接,即RealConnection 的创建。
            RealConnection resultConnection = findHealthyConnection(connectTimeout, readTimeout,
                    writeTimeout, pingIntervalMillis, connectionRetryEnabled, doExtensiveHealthChecks);
            //根据HTTP协议的版本创建Http1Codec或Http2Codec
            HttpCodec resultCodec = resultConnection.newCodec(client, chain, this);

            synchronized (connectionPool) {
                codec = resultCodec;
                return resultCodec;
            }
        } catch (IOException e) {
            throw new RouteException(e);
        }
    }

1.1StreamAllocation.newStream()调用findConnection()方法来建立连接。

最终调用Java里的套接字Socket里的connect()方法
调用顺序
StreamAllocation.newStream()-->findHealthyConnection()-->findConnection()
findConnection()中的流程如下:

  • 1.查找是否有完整的连接可用:
    Socket没有关闭

输入流没有关闭
输出流没有关闭
Http2连接没有关闭

  • 2.连接池中是否有可用的连接,如果有则可用。
  • 3.如果没有可用连接,则自己创建一个。
  • 4.开始TCP连接以及TLS握手操作。(握手操作调用RealConnection.connect()-->最终调用ava里的套接字Socket里的connect()方法)
  • 5.将新创建的连接加入连接池。
public final class StreamAllocation {
    /**
     * Returns a connection to host a new stream. This prefers the existing connection if it exists,
     * then the pool, finally building a new connection.
     * 返回一个连接来托管一个新的流。 这更喜欢现有的连接(如果存在的话),然后是池,最后建立一个新的连接。
     */
    private RealConnection findConnection(int connectTimeout, int readTimeout, int writeTimeout,
                                          int pingIntervalMillis, boolean connectionRetryEnabled) throws IOException {
        boolean foundPooledConnection = false;
        RealConnection result = null;
        Route selectedRoute = null;
        Connection releasedConnection;
        Socket toClose;
        synchronized (connectionPool) {
            if (released) throw new IllegalStateException("released");
            if (codec != null) throw new IllegalStateException("codec != null");
            if (canceled) throw new IOException("Canceled");

            // Attempt to use an already-allocated connection. We need to be careful here because our
            // already-allocated connection may have been restricted from creating new streams.
            //尝试使用已分配的连接。 我们在这里需要小心,因为我们已经分配的连接可能已经被限制在创建新的流中。
            releasedConnection = this.connection;
            toClose = releaseIfNoNewStreams();
            //1 查看是否有完好的连接
            if (this.connection != null) {
                // We had an already-allocated connection and it's good.
                result = this.connection;
                releasedConnection = null;
            }
            if (!reportedAcquired) {
                // If the connection was never reported acquired, don't report it as released!
                releasedConnection = null;
            }
            //2 连接池中是否用可用的连接,有则使用
            if (result == null) {
                // Attempt to get a connection from the pool.
                Internal.instance.get(connectionPool, address, this, null);
                if (connection != null) {
                    foundPooledConnection = true;
                    result = connection;
                } else {
                    selectedRoute = route;
                }
            }
        }
        closeQuietly(toClose);

        if (releasedConnection != null) {
            eventListener.connectionReleased(call, releasedConnection);
        }
        if (foundPooledConnection) {
            eventListener.connectionAcquired(call, result);
        }
        if (result != null) {
            // If we found an already-allocated or pooled connection, we're done.
            //如果我们找到了已经分配或者连接的连接,我们就完成了。
            return result;
        }

        // If we need a route selection, make one. This is a blocking operation.
        //如果我们需要路线选择,请选择一个。 这是一项阻止操作。
        //线程的选择,多IP操作
        boolean newRouteSelection = false;
        if (selectedRoute == null && (routeSelection == null || !routeSelection.hasNext())) {
            newRouteSelection = true;
            routeSelection = routeSelector.next();
        }
        //3 如果没有可用连接,则自己创建一个
        synchronized (connectionPool) {
            if (canceled) throw new IOException("Canceled");

            if (newRouteSelection) {
                // Now that we have a set of IP addresses, make another attempt at getting a connection from
                // the pool. This could match due to connection coalescing.
               // 现在我们有一组IP地址,再次尝试从池中获取连接。 这可能由于连接合并而匹配。
                List<Route> routes = routeSelection.getAll();
                for (int i = 0, size = routes.size(); i < size; i++) {
                    Route route = routes.get(i);
                    Internal.instance.get(connectionPool, address, this, route);
                    if (connection != null) {
                        foundPooledConnection = true;
                        result = connection;
                        this.route = route;
                        break;
                    }
                }
            }

            if (!foundPooledConnection) {
                if (selectedRoute == null) {
                    selectedRoute = routeSelection.next();
                }

                // Create a connection and assign it to this allocation immediately. This makes it possible
                // for an asynchronous cancel() to interrupt the handshake we're about to do.
                //创建一个连接并立即将其分配给该分配。 这使得异步cancel()可以中断我们即将进行的握手。
                route = selectedRoute;
                refusedStreamCount = 0;
                result = new RealConnection(connectionPool, selectedRoute);
                acquire(result, false);
            }
        }

        // If we found a pooled connection on the 2nd time around, we're done.
        //如果我们第二次发现一个连接池,我们就完成了。
        if (foundPooledConnection) {
            eventListener.connectionAcquired(call, result);
            return result;
        }

        // Do TCP + TLS handshakes. This is a blocking operation.
        //4 开始TCP以及TLS握手操作,这是阻塞操作
        result.connect(connectTimeout, readTimeout, writeTimeout, pingIntervalMillis,
                connectionRetryEnabled, call, eventListener);
        routeDatabase().connected(result.route());

        //5 将新创建的连接,放在连接池中
        Socket socket = null;
        synchronized (connectionPool) {
            reportedAcquired = true;

            // Pool the connection.
            Internal.instance.put(connectionPool, result);

            // If another multiplexed connection to the same address was created concurrently, then
            // release this connection and acquire that one.
            //如果同时创建了到同一地址的另一个多路复用连接,则释放此连接并获取该连接。
            if (result.isMultiplexed()) {
                socket = Internal.instance.deduplicate(connectionPool, address, this);
                result = connection;
            }
        }
        closeQuietly(socket);

        eventListener.connectionAcquired(call, result);
        return result;
    }
}

1.2 调用该方法的RealConnection.connect()方法建立连接

  • connect-->connectSocket()进行socket连接-->Platform.get().connectSocket()-->socket.connect(address, connectTimeout);(此时进行了三次握手)
  • 握手完成后调用establishProtocol()

注:
当客户端调用connect时,触发了连接请求,向服务器发送了SYN J包,这时connect进入阻塞状态;服务器监听到连接请求,即收到SYN J包,调用accept函 数接收请求向客户端发送SYN K ,ACK J+1,这时accept进入阻塞状态;客户端收到服务器的SYN K ,ACK J+1之后,这时connect返回,并对SYN K进行确认;服务器收到ACK K+1时,accept返回,至此三次握手完毕,连接建立。

connect()

    public void connect(int connectTimeout, int readTimeout, int writeTimeout,
                        int pingIntervalMillis, boolean connectionRetryEnabled, Call call,
                        EventListener eventListener) {
        if (protocol != null) throw new IllegalStateException("already connected");
        //线路选择
        RouteException routeException = null;
        List<ConnectionSpec> connectionSpecs = route.address().connectionSpecs();
        ConnectionSpecSelector connectionSpecSelector = new ConnectionSpecSelector(connectionSpecs);

        if (route.address().sslSocketFactory() == null) {
            if (!connectionSpecs.contains(ConnectionSpec.CLEARTEXT)) {
                throw new RouteException(new UnknownServiceException(
                        "CLEARTEXT communication not enabled for client"));
            }
            String host = route.address().url().host();
            if (!Platform.get().isCleartextTrafficPermitted(host)) {
                throw new RouteException(new UnknownServiceException(
                        "CLEARTEXT communication to " + host + " not permitted by network security policy"));
            }
        }
        //开始连接
        while (true) {
            try {
                //建立隧道连接
                if (route.requiresTunnel()) {
                    connectTunnel(connectTimeout, readTimeout, writeTimeout, call, eventListener);
                    if (rawSocket == null) {
                        // We were unable to connect the tunnel but properly closed down our resources.
                        //我们无法连接隧道,但正确关闭了我们的资源。
                        break;
                    }
                } else {
                    //建立普通连接
                    connectSocket(connectTimeout, readTimeout, call, eventListener);
                }
                // 建立协议
                //不管是建立隧道连接,还是建立普通连接,都少不了 建立协议 这一步。
                // 这一步是在建立好了TCP连接之后,而在该TCP能被拿来收发数据之前执行的。
                // 它主要为数据的加密传输做一些初始化,比如TLS握手,HTTP/2的协议协商等
                establishProtocol(connectionSpecSelector, pingIntervalMillis, call, eventListener);
                eventListener.connectEnd(call, route.socketAddress(), route.proxy(), protocol);
                break;
                //完成连接
            } catch (IOException e) {
                closeQuietly(socket);
                closeQuietly(rawSocket);
                socket = null;
                rawSocket = null;
                source = null;
                sink = null;
                handshake = null;
                protocol = null;
                http2Connection = null;

                eventListener.connectFailed(call, route.socketAddress(), route.proxy(), null, e);

                if (routeException == null) {
                    routeException = new RouteException(e);
                } else {
                    routeException.addConnectException(e);
                }

                if (!connectionRetryEnabled || !connectionSpecSelector.connectionFailed(e)) {
                    throw routeException;
                }
            }
        }

        if (route.requiresTunnel() && rawSocket == null) {
            ProtocolException exception = new ProtocolException("Too many tunnel connections attempted: "
                    + MAX_TUNNEL_ATTEMPTS);
            throw new RouteException(exception);
        }

        if (http2Connection != null) {
            synchronized (connectionPool) {
                allocationLimit = http2Connection.maxConcurrentStreams();
            }
        }
    }

connectSocket()

    /**
     * Does all the work necessary to build a full HTTP or HTTPS connection on a raw socket.
     * 完成在原始套接字上构建完整的HTTP或HTTPS连接所需的所有工作。
     */
    private void connectSocket(int connectTimeout, int readTimeout, Call call,
                               EventListener eventListener) throws IOException {
        Proxy proxy = route.proxy();
        Address address = route.address();
        //根据代理类型的不同处理Socket
        rawSocket = proxy.type() == Proxy.Type.DIRECT || proxy.type() == Proxy.Type.HTTP
                ? address.socketFactory().createSocket()
                : new Socket(proxy);

        eventListener.connectStart(call, route.socketAddress(), proxy);
        rawSocket.setSoTimeout(readTimeout);
        try {
            //建立Socket连接
            Platform.get().connectSocket(rawSocket, route.socketAddress(), connectTimeout);
        } catch (ConnectException e) {
            ConnectException ce = new ConnectException("Failed to connect to " + route.socketAddress());
            ce.initCause(e);
            throw ce;
        }

        // The following try/catch block is a pseudo hacky way to get around a crash on Android 7.0
        // More details:
        // https://github.com/square/okhttp/issues/3245
        // https://android-review.googlesource.com/#/c/271775/
        try {
            //获取输入/输出流
            source = Okio.buffer(Okio.source(rawSocket));
            sink = Okio.buffer(Okio.sink(rawSocket));
        } catch (NullPointerException npe) {
            if (NPE_THROW_WITH_NULL.equals(npe.getMessage())) {
                throw new IOException(npe);
            }
        }
    }

public class Platform {
  public void connectSocket(Socket socket, InetSocketAddress address,
      int connectTimeout) throws IOException {
    //最终调用java的connect
    socket.connect(address, connectTimeout);
  }
}

connectTunnel()隧道链接

    /**
     * Does all the work to build an HTTPS connection over a proxy tunnel. The catch here is that a
     * proxy server can issue an auth challenge and then close the connection.
     * 是否通过代理隧道建立HTTPS连接的所有工作。 这里的问题是代理服务器可以发出一个验证质询,然后关闭连接。
     */
    private void connectTunnel(int connectTimeout, int readTimeout, int writeTimeout, Call call,
                               EventListener eventListener) throws IOException {
        //1-构造一个 建立隧道连接 请求。
        Request tunnelRequest = createTunnelRequest();
        HttpUrl url = tunnelRequest.url();
        for (int i = 0; i < MAX_TUNNEL_ATTEMPTS; i++) {
            //2 与HTTP代理服务器建立TCP连接。
            connectSocket(connectTimeout, readTimeout, call, eventListener);
            //3 创建隧道。这主要是将 建立隧道连接 请求发送给HTTP代理服务器,并处理它的响应
            tunnelRequest = createTunnel(readTimeout, writeTimeout, tunnelRequest, url);

            if (tunnelRequest == null) break; // Tunnel successfully created.

            // The proxy decided to close the connection after an auth challenge. We need to create a new
            // connection, but this time with the auth credentials.
            closeQuietly(rawSocket);
            rawSocket = null;
            sink = null;
            source = null;
            eventListener.connectEnd(call, route.socketAddress(), route.proxy(), null);
            //重复上面的第2和第3步,直到建立好了隧道连接。
        }
    }

createTunnel()

public final class RealConnection extends Http2Connection.Listener implements Connection {
    /**
     * To make an HTTPS connection over an HTTP proxy, send an unencrypted CONNECT request to create
     * the proxy connection. This may need to be retried if the proxy requires authorization.
     * 要通过HTTP代理建立HTTPS连接,请发送未加密的CONNECT请求以创建代理连接。 如果代理需要授权,则可能需要重试。
     */
    private Request createTunnel(int readTimeout, int writeTimeout, Request tunnelRequest,
                                 HttpUrl url) throws IOException {
        // Make an SSL Tunnel on the first message pair of each SSL + proxy connection.
        // 在每个SSL +代理连接的第一个消息对上创建一个SSL隧道。
        String requestLine = "CONNECT " + Util.hostHeader(url, true) + " HTTP/1.1";
        while (true) {
            Http1Codec tunnelConnection = new Http1Codec(null, null, source, sink);
            source.timeout().timeout(readTimeout, MILLISECONDS);
            sink.timeout().timeout(writeTimeout, MILLISECONDS);
            tunnelConnection.writeRequest(tunnelRequest.headers(), requestLine);
            //sink.flush();
            tunnelConnection.finishRequest();
            Response response = tunnelConnection.readResponseHeaders(false)
                    .request(tunnelRequest)
                    .build();
            // The response body from a CONNECT should be empty, but if it is not then we should consume
            // it before proceeding.
            long contentLength = HttpHeaders.contentLength(response);
            if (contentLength == -1L) {
                contentLength = 0L;
            }
            Source body = tunnelConnection.newFixedLengthSource(contentLength);
            Util.skipAll(body, Integer.MAX_VALUE, TimeUnit.MILLISECONDS);
            body.close();

            switch (response.code()) {
                case HTTP_OK:
                    // Assume the server won't send a TLS ServerHello until we send a TLS ClientHello. If
                    // that happens, then we will have buffered bytes that are needed by the SSLSocket!
                    // This check is imperfect: it doesn't tell us whether a handshake will succeed, just
                    // that it will almost certainly fail because the proxy has sent unexpected data.
                    if (!source.buffer().exhausted() || !sink.buffer().exhausted()) {
                        throw new IOException("TLS tunnel buffered too many bytes!");
                    }
                    return null;

                case HTTP_PROXY_AUTH:
                    tunnelRequest = route.address().proxyAuthenticator().authenticate(route, response);
                    if (tunnelRequest == null)
                        throw new IOException("Failed to authenticate with proxy");

                    if ("close".equalsIgnoreCase(response.header("Connection"))) {
                        return tunnelRequest;
                    }
                    break;

                default:
                    throw new IOException(
                            "Unexpected response code for CONNECT: " + response.code());
            }
        }
    }
}

猜你喜欢

转载自blog.csdn.net/weixin_34152820/article/details/87190271