ConnectInterceptor
ConnectInterceptor 即连接拦截器，此拦截器即和 Request 请求的拦截相关；
查看 ConnectInterceptor的 intercept 方法；
@Override public Response intercept(Chain chain) throws IOException {
RealInterceptorChain realChain = (RealInterceptorChain) chain;
Request request = realChain.request(); // 拿到 Request 请求
StreamAllocation streamAllocation = realChain.streamAllocation();// 获取 streamAllocation 实例，此实例在 RetryAndFollowUpInterceptor 进行创建过
// We need the network to satisfy this request. Possibly for validating a conditional GET.
boolean doExtensiveHealthChecks = !request.method().equals("GET");// 是否是 get 方法
HttpCodec httpCodec = streamAllocation.newStream(client, chain, doExtensiveHealthChecks); // 获取 HttpCodec 实例
RealConnection connection = streamAllocation.connection(); // 拿到 RealConnection 实例
return realChain.proceed(request, streamAllocation, httpCodec, connection);// 调用下一个拦截器
}
ConnectInterceptor 获取到 RetryAndFollowUpInterceptor 拦截器中创建的 StreamAllocation 实例，通过streamAllocation，获取到 HttpCodec 和 RealConnection 实例对象，最后传递相关对象参数调用到下一个拦截器；
获取 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 resultConnection = findHealthyConnection(connectTimeout, readTimeout,
writeTimeout, pingIntervalMillis, connectionRetryEnabled, doExtensiveHealthChecks);
HttpCodec resultCodec = resultConnection.newCodec(client, chain, this);
synchronized (connectionPool) {
codec = resultCodec;
return resultCodec;
}
} catch (IOException e) {
throw new RouteException(e);
}
}
newStream 通过 findHealthyConnection 方法获取到 RealConnection 实例对象，同时创建 HttpCodec 实例对象；
private RealConnection findHealthyConnection(int connectTimeout, int readTimeout,
int writeTimeout, int pingIntervalMillis, boolean connectionRetryEnabled,
boolean doExtensiveHealthChecks) throws IOException {
while (true) {
RealConnection candidate = findConnection(connectTimeout, readTimeout, writeTimeout,
pingIntervalMillis, connectionRetryEnabled); // 查找 RealConnection
// If this is a brand new connection, we can skip the extensive health checks.
synchronized (connectionPool) {
if (candidate.successCount == 0) {
return candidate;
}
}
// Do a (potentially slow) check to confirm that the pooled connection is still good. If it
// isn't, take it out of the pool and start again.
if (!candidate.isHealthy(doExtensiveHealthChecks)) {
noNewStreams();
continue;
}
return candidate;
}
}
继续调用 findConnection 方法获取 RealConnection 实例对象；
/**
* Returns a connection to host a new stream. This prefers the existing connection if it exists,
* then the pool, finally building a new connection.
* 返回 RealConnection 如果存在 connection 返回，否则重新创建一个 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.
// 试图去使用已经存在 connection
releasedConnection = this.connection;
toClose = releaseIfNoNewStreams();
if (this.connection != null) { // 不为空，使用 connection
// 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;
}
if (result == null) { // 试图 从 connectionPool 获取一个 connection
// 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) { // 使用到了 connectionPool 中的 connection
// 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.
// 创建路由，从新路由中去获取 connection
boolean newRouteSelection = false;
if (selectedRoute == null && (routeSelection == null || !routeSelection.hasNext())) {
newRouteSelection = true;
routeSelection = routeSelector.next();
}
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.
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) { // 没有从 connectionPool 获取到 connection
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.
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) { // 获取到了 connection
eventListener.connectionAcquired(call, result);
return result;
}
// Do TCP + TLS handshakes. This is a blocking operation.
result.connect(connectTimeout, readTimeout, writeTimeout, pingIntervalMillis,
connectionRetryEnabled, call, eventListener);进行连接
routeDatabase().connected(result.route()); //
Socket socket = null;
synchronized (connectionPool) {
reportedAcquired = true;
// Pool the connection.
Internal.instance.put(connectionPool, result); // connection 存入 connectionPool 中
// 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;
}
findConnection 方法主要做了以下事情；
先获取定义的 connection，为空的话，通过 connectionPool 连接池中获取，还为空的话，更换路由去查找，查找不到最后创建 RealConnection 放入到下一个路由以及 connectionPool 中；
ConnectionPool
public final class ConnectionPool {
private final Deque<RealConnection> connections = new ArrayDeque<>(); // RealConnection 队列
final RouteDatabase routeDatabase = new RouteDatabase();
boolean cleanupRunning;
private static final Executor executor = new ThreadPoolExecutor(0 /* corePoolSize */,
Integer.MAX_VALUE /* maximumPoolSize */, 60L /* keepAliveTime */, TimeUnit.SECONDS,
new SynchronousQueue<Runnable>(), Util.threadFactory("OkHttp ConnectionPool", true));
void put(RealConnection connection) {
assert (Thread.holdsLock(this));
if (!cleanupRunning) {
cleanupRunning = true;
executor.execute(cleanupRunnable); // 定期执行清除
}
connections.add(connection);
}
@Nullable RealConnection get(Address address, StreamAllocation streamAllocation, Route route) {
assert (Thread.holdsLock(this));
for (RealConnection connection : connections) { // 获取 connection
if (connection.isEligible(address, route)) {
streamAllocation.acquire(connection, true);
return connection;
}
}
return null;
}
ConnectionPool 维护了 RealConnection 的双向队列，同时当添加 RealConnection 的时候会开启线程，对 connectionPool 进行相关的清理操作；
ConnectionPool 的 get 方法会对比参数的 Host 对比成功，返回对应 connection，即 Host 相同的时候复用 connection；
public boolean isEligible(Address address, @Nullable Route route) {
// If this connection is not accepting new streams, we're done.
if (allocations.size() >= allocationLimit || noNewStreams) return false;
// If the non-host fields of the address don't overlap, we're done.
if (!Internal.instance.equalsNonHost(this.route.address(), address)) return false;
// If the host exactly matches, we're done: this connection can carry the address.
if (address.url().host().equals(this.route().address().url().host())) { // 两者Host 相同
return true; // This connection is a perfect match.
}
StreamAllocation 的 acquire 方法则会赋值 connection 变量；
public void acquire(RealConnection connection, boolean reportedAcquired) {
assert (Thread.holdsLock(connectionPool));
if (this.connection != null) throw new IllegalStateException();
this.connection = connection;
this.reportedAcquired = reportedAcquired;
connection.allocations.add(new StreamAllocationReference(this, callStackTrace));
}
分析到这里，我们知道了 ConnectInterceptor 会根据 Reqsuet 请求的 Url 的 Host 去对比 ConnectionPool 中查找是否有对应的 connetction 连接已经存在，如果有即复用 connetction ；这里复用的规则即为相同 Host 的请求，这样避免了重新创建 connection；
获取到了 RealConnection 之后即进行进行 connect；
RealConnection
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) { // 无ssl
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 {// 创建 socket 连接 一般走这里
connectSocket(connectTimeout, readTimeout, call, eventListener);
}
establishProtocol(connectionSpecSelector, pingIntervalMillis, call, eventListener); // 建立协议，一般为TLS握手协议
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();
}
}
}
connect 方法主要执行以下流程；
1 通过路由获取安全套件，并验证安全套件是否和协议一致：对于HTTP协议的请求，安全套件中必须包含CLEARTEXT，CLEATTEXT代表着明文传输；Android平台本身的安全策略是否允许向相应的主机发送明文请求。
2 进入循环创建连接直到创建成功，跳出循环。
3 首先根据路由判断是否需要建立隧道 ，建立隧道连接 或者建立普通的连接
4 建立协议，指的是建立TSL握手协议
5 对于HTTP2协议，设置连接的最大分配数，指一条HTTP连接上最多同时存在的请求数目。
private void establishProtocol(ConnectionSpecSelector connectionSpecSelector,
int pingIntervalMillis, Call call, EventListener eventListener) throws IOException {
if (route.address().sslSocketFactory() == null) { // http
protocol = Protocol.HTTP_1_1;
socket = rawSocket;
return;
}
eventListener.secureConnectStart(call);
connectTls(connectionSpecSelector);// TLS协议握手过程 包括TLS 握手过程
eventListener.secureConnectEnd(call, handshake);
if (protocol == Protocol.HTTP_2) { // https
socket.setSoTimeout(0); // HTTP/2 connection timeouts are set per-stream.
http2Connection = new Http2Connection.Builder(true)
.socket(socket, route.address().url().host(), source, sink)
.listener(this)
.pingIntervalMillis(pingIntervalMillis)
.build();
http2Connection.start();
}
}
如果是HTTP协议，不需要建立协议的过程，此时TCP握手已经完成，可以在这个连接上开始于服务器的通信；如果是HTTPS、HTTP2 协议则还需要建立协议 TLS协议，完成TLS的握手，验证服务器证书，以及协商机密算法、传输秘钥。
到这里即完成了 connect 过程分析，返回 RealConnection；
当拿到了 RealConnection 之后，创建 HttpCodec ;
public HttpCodec newCodec(OkHttpClient client, Interceptor.Chain chain,
StreamAllocation streamAllocation) throws SocketException {
if (http2Connection != null) {
return new Http2Codec(client, chain, streamAllocation, http2Connection);
} else {
socket.setSoTimeout(chain.readTimeoutMillis());
source.timeout().timeout(chain.readTimeoutMillis(), MILLISECONDS);
sink.timeout().timeout(chain.writeTimeoutMillis(), MILLISECONDS);
return new Http1Codec(client, streamAllocation, source, sink);
}
}
HttpCodec
public Http2Codec(OkHttpClient client, Interceptor.Chain chain, StreamAllocation streamAllocation,
Http2Connection connection) {
this.client = client;
this.chain = chain;
this.streamAllocation = streamAllocation;
this.connection = connection;
}
Http2Codec 封装请求的相关参数；
RealConnection 和 Http2Codec 作为参数传入下一个拦截器；
网友评论