OkHttp是google推薦的優(yōu)秀網(wǎng)絡(luò)請求框架篮绿，深受開發(fā)者青睞，它的主要特點如下：
1.通過socket來實現(xiàn)網(wǎng)絡(luò)請求
2.支持http1.1 http2.0
3.連接復(fù)用寡痰，連接池緩存所有連接，自動清除空閑無用的連接，復(fù)用可用連接中剩，減少連接耗時
4.通過攔截器的方式來實現(xiàn)，每個攔截器處理自己的事情抒寂，結(jié)構(gòu)清晰
基本的使用方式如下
String url = "http://wwww.baidu.com";
OkHttpClient okHttpClient = new OkHttpClient();
final Request request = new Request.Builder()
.url(url)
.build();
Call call = okHttpClient.newCall(request);
call.enqueue(new Callback() {
@Override
public void onFailure(Call call, IOException e) {
}
@Override
public void onResponse(Call call, Response response) throws IOException {
}
});
我們先看一下OkHttpClient
public class OkHttpClient implements Cloneable, Call.Factory, WebSocket.Factory {
public OkHttpClient() {
this(new Builder());
}
OkHttpClient(Builder builder) {
//這里是將builder里面的值賦值過來
}
}
public static final class Builder {
//線程分發(fā)器结啼，保存有正在執(zhí)行和緩存的請求隊列，通過ExecutorService線程池屈芜，進行處理
Dispatcher dispatcher;
@Nullable Proxy proxy;
//連接的協(xié)議郊愧，常有的有3種朴译，http1.0 http1.1 http2.0
//http1.0
//連接后發(fā)起請求，收到響應(yīng)后關(guān)閉連接属铁，沒起請求都得重新建立連接
//http1.1
//連接復(fù)用眠寿，收到響應(yīng)后，可以不關(guān)閉連接红选，繼續(xù)發(fā)請求
//keep-alive頭表示不關(guān)閉澜公，close頭表示關(guān)閉
//管道機制，客戶端發(fā)請求不用等服務(wù)器返回喇肋，但服務(wù)器必須順序返回
//添加100 continue的狀態(tài)碼坟乾，傳輸大文件時，先問服務(wù)器當前能否處理蝶防，能處理再發(fā)body甚侣，節(jié)省帶寬
//引入Chunked transfer-coding進行分塊傳輸，避免緩沖整個消息帶來的負載间学，產(chǎn)生一塊數(shù)據(jù)后就發(fā)送
//添加range頭殷费，支持斷點續(xù)傳
//請求頭必須帶上host，解決虛擬主機的問題
//http2.0
//支持多路復(fù)用低葫，可同時發(fā)請求详羡，請求可以設(shè)置優(yōu)先級，優(yōu)先處理優(yōu)先級高的
//頭壓縮嘿悬，維護了一個字段实柠，增量更新頭信息
//支持推送
//內(nèi)容采用二進制
List<Protocol> protocols;
List<ConnectionSpec> connectionSpecs;
final List<Interceptor> interceptors = new ArrayList<>();
final List<Interceptor> networkInterceptors = new ArrayList<>();
EventListener.Factory eventListenerFactory;
ProxySelector proxySelector;
CookieJar cookieJar;
@Nullable Cache cache;
@Nullable InternalCache internalCache;
SocketFactory socketFactory;
@Nullable SSLSocketFactory sslSocketFactory;
@Nullable CertificateChainCleaner certificateChainCleaner;
HostnameVerifier hostnameVerifier;
CertificatePinner certificatePinner;
Authenticator proxyAuthenticator;
Authenticator authenticator;
//連接池，管理連接
ConnectionPool connectionPool;
Dns dns;
boolean followSslRedirects;
boolean followRedirects;
boolean retryOnConnectionFailure;
int connectTimeout;
int readTimeout;
int writeTimeout;
int pingInterval;
public Builder() {
dispatcher = new Dispatcher();
protocols = DEFAULT_PROTOCOLS;
connectionSpecs = DEFAULT_CONNECTION_SPECS;
eventListenerFactory = EventListener.factory(EventListener.NONE);
proxySelector = ProxySelector.getDefault();
cookieJar = CookieJar.NO_COOKIES;
socketFactory = SocketFactory.getDefault();
hostnameVerifier = OkHostnameVerifier.INSTANCE;
certificatePinner = CertificatePinner.DEFAULT;
proxyAuthenticator = Authenticator.NONE;
authenticator = Authenticator.NONE;
connectionPool = new ConnectionPool();
dns = Dns.SYSTEM;
followSslRedirects = true;
followRedirects = true;
retryOnConnectionFailure = true;
connectTimeout = 10_000;
readTimeout = 10_000;
writeTimeout = 10_000;
pingInterval = 0;
}
}
}
看一下ConnectionPool連接池的實現(xiàn)
public final class ConnectionPool {
//線程池善涨，處理連接的定期清理
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;
//執(zhí)行任務(wù)窒盐，清除閑置的連接
executor.execute(cleanupRunnable);
}
connections.add(connection);
}
private final Runnable cleanupRunnable = new Runnable() {
@Override public void run() {
while (true) {
//獲取下一次執(zhí)行的時間
long waitNanos = cleanup(System.nanoTime());
if (waitNanos == -1) return;
if (waitNanos > 0) {
long waitMillis = waitNanos / 1000000L;
waitNanos -= (waitMillis * 1000000L);
synchronized (ConnectionPool.this) {
try {
ConnectionPool.this.wait(waitMillis, (int) waitNanos);
} catch (InterruptedException ignored) {
}
}
}
}
}
};
//清除閑置的連接
long cleanup(long now) {
int inUseConnectionCount = 0;
int idleConnectionCount = 0;
RealConnection longestIdleConnection = null;
long longestIdleDurationNs = Long.MIN_VALUE;
//這個循環(huán)的作用是找出最長閑置的連接
synchronized (this) {
for (Iterator<RealConnection> i = connections.iterator(); i.hasNext(); ) {
RealConnection connection = i.next();
//返回這個連接上面流空間的數(shù)量，為0的話钢拧，就是閑置
if (pruneAndGetAllocationCount(connection, now) > 0) {
inUseConnectionCount++;
continue;
}
idleConnectionCount++;
// 獲得閑置持續(xù)的時間
long idleDurationNs = now - connection.idleAtNanos;
//這里是記錄最長閑置時間
if (idleDurationNs > longestIdleDurationNs) {
longestIdleDurationNs = idleDurationNs;
longestIdleConnection = connection;
}
}
//如果最長閑置時間大于一定值蟹漓，或者閑置連接數(shù)量大于最大值，就會移除最長閑置的連接
if (longestIdleDurationNs >= this.keepAliveDurationNs
|| idleConnectionCount > this.maxIdleConnections) {
connections.remove(longestIdleConnection);
} else if (idleConnectionCount > 0) {
// keepAliveDurationNs表示允許閑置的時間源内，longestIdleDurationNs是當前最長閑置時間葡粒，差值是下次檢查的時間，過了這個時間來檢查膜钓，剛好該連接過期
return keepAliveDurationNs - longestIdleDurationNs;
} else if (inUseConnectionCount > 0) {
return keepAliveDurationNs;
} else {
// No connections, idle or in use.
cleanupRunning = false;
return -1;
}
}
closeQuietly(longestIdleConnection.socket());
return 0;
}
//返回這個連接上面流的數(shù)量
private int pruneAndGetAllocationCount(RealConnection connection, long now) {
List<Reference<StreamAllocation>> references = connection.allocations;
for (int i = 0; i < references.size(); ) {
Reference<StreamAllocation> reference = references.get(i);
if (reference.get() != null) {
i++;
continue;
}
//一個連接上面塔鳍，出現(xiàn)來空的流
StreamAllocation.StreamAllocationReference streamAllocRef =
(StreamAllocation.StreamAllocationReference) reference;
String message = "A connection to " + connection.route().address().url()
+ " was leaked. Did you forget to close a response body?";
Platform.get().logCloseableLeak(message, streamAllocRef.callStackTrace);
references.remove(i);
connection.noNewStreams = true;
// If this was the last allocation, the connection is eligible for immediate eviction.
if (references.isEmpty()) {
//記錄這個連接空閑的時間
connection.idleAtNanos = now - keepAliveDurationNs;
return 0;
}
}
return references.size();
}
}
連接池是用來管理所有連接，當某個連接空閑時長超過一定值呻此，或者空閑的連接數(shù)量超過一定值，就會清除連接空閑時間最長的那個
看一下
final Request request = new Request.Builder() .url(url).build();
public final class Request {
Request(Builder builder) {
this.url = builder.url;
this.method = builder.method;
this.headers = builder.headers.build();
this.body = builder.body;
this.tag = builder.tag != null ? builder.tag : this;
}
public Builder url(String url) {
HttpUrl parsed = HttpUrl.parse(url);
if (parsed == null) throw new IllegalArgumentException("unexpected url: " + url);
return url(parsed);
}
public static class Builder {
public Builder url(HttpUrl url) {
if (url == null) throw new NullPointerException("url == null");
this.url = url;
return this;
}
public Request build() {
if (url == null) throw new IllegalStateException("url == null");
return new Request(this);
}
}
}
這里只是解析了url腔寡，將url封裝到request中焚鲜。
看一下 Call call = okHttpClient.newCall(request)和call方法
public class OkHttpClient implements Cloneable, Call.Factory, WebSocket.Factory {
public Call newCall(Request request) {
return RealCall.newRealCall(this, request, false /* for web socket */);
}
}
final class RealCall implements Call {
private RealCall(OkHttpClient client, Request originalRequest, boolean forWebSocket) {
this.client = client;
this.originalRequest = originalRequest;
this.forWebSocket = forWebSocket;
//這里定義了一個重試攔截器
this.retryAndFollowUpInterceptor = new RetryAndFollowUpInterceptor(client, forWebSocket);
}
static RealCall newRealCall(OkHttpClient client, Request originalRequest, boolean forWebSocket) {
// Safely publish the Call instance to the EventListener.
RealCall call = new RealCall(client, originalRequest, forWebSocket);
call.eventListener = client.eventListenerFactory().create(call);
return call;
}
//這里開始執(zhí)行
public void enqueue(Callback responseCallback) {
synchronized (this) {
if (executed) throw new IllegalStateException("Already Executed");
executed = true;
}
captureCallStackTrace();
eventListener.callStart(this);
//之前在OkHttpClient初始化的時候，創(chuàng)建了線程分發(fā)器
client.dispatcher().enqueue(new AsyncCall(responseCallback));
}
final class AsyncCall extends NamedRunnable {
private final Callback responseCallback;
@Override protected void execute() {
boolean signalledCallback = false;
try {
//通過攔截器鏈獲得最后的響應(yīng)
Response response = getResponseWithInterceptorChain();
if (retryAndFollowUpInterceptor.isCanceled()) {
signalledCallback = true;
responseCallback.onFailure(RealCall.this, new IOException("Canceled"));
} else {
signalledCallback = true;
responseCallback.onResponse(RealCall.this, response);
}
} catch (IOException e) {
if (signalledCallback) {
// Do not signal the callback twice!
Platform.get().log(INFO, "Callback failure for " + toLoggableString(), e);
} else {
eventListener.callFailed(RealCall.this, e);
responseCallback.onFailure(RealCall.this, e);
}
} finally {
client.dispatcher().finished(this);
}
}
}
Response getResponseWithInterceptorChain() throws IOException {
// Build a full stack of interceptors.
List<Interceptor> interceptors = new ArrayList<>();
//用戶自己的攔截器
interceptors.addAll(client.interceptors());
//重試攔截器，請求發(fā)生了錯誤忿磅，會重試
interceptors.add(retryAndFollowUpInterceptor);
//橋接攔截器
interceptors.add(new BridgeInterceptor(client.cookieJar()));
//緩存攔截器
interceptors.add(new CacheInterceptor(client.internalCache()));
//連接攔截器
interceptors.add(new ConnectInterceptor(client));
if (!forWebSocket) {
interceptors.addAll(client.networkInterceptors());
}
//請求攔截器
interceptors.add(new CallServerInterceptor(forWebSocket));
//將攔截器放到interceptors中
Interceptor.Chain chain = new RealInterceptorChain(interceptors, null, null, null, 0,
originalRequest, this, eventListener, client.connectTimeoutMillis(),
client.readTimeoutMillis(), client.writeTimeoutMillis());
return chain.proceed(originalRequest);
}
}
看一下攔截器是如何攔截的
public final class RealInterceptorChain implements Interceptor.Chain {
public RealInterceptorChain(List<Interceptor> interceptors, StreamAllocation streamAllocation,
HttpCodec httpCodec, RealConnection connection, int index, Request request, Call call,
EventListener eventListener, int connectTimeout, int readTimeout, int writeTimeout) {
//攔截器的列表
this.interceptors = interceptors;
//連接
this.connection = connection;
//流空間
this.streamAllocation = streamAllocation;
//http解碼器
this.httpCodec = httpCodec;
//當前該調(diào)用攔截器的索引
this.index = index;
this.request = request;
this.call = call;
this.eventListener = eventListener;
this.connectTimeout = connectTimeout;
this.readTimeout = readTimeout;
this.writeTimeout = writeTimeout;
}
//開始執(zhí)行
public Response proceed(Request request, StreamAllocation streamAllocation, HttpCodec httpCodec,
RealConnection connection) throws IOException {
//index表示當前要執(zhí)行的攔截器
if (index >= interceptors.size()) throw new AssertionError();
//表示執(zhí)行的次數(shù)
calls++;
// 調(diào)用下一個攔截器
RealInterceptorChain next = new RealInterceptorChain(interceptors, streamAllocation, httpCodec,
connection, index + 1, request, call, eventListener, connectTimeout, readTimeout,
writeTimeout);
Interceptor interceptor = interceptors.get(index);
//調(diào)用攔截器進行處理
Response response = interceptor.intercept(next);
return response;
}
}
看一下第一個攔截器RetryAndFollowUpInterceptor
public final class RetryAndFollowUpInterceptor implements Interceptor {
public Response intercept(Chain chain) throws IOException {
Request request = chain.request();
RealInterceptorChain realChain = (RealInterceptorChain) chain;
Call call = realChain.call();
EventListener eventListener = realChain.eventListener();
//這里新建了流空間糯彬，里面有連接池，請求的地址等信息葱她，會將該流空間往下傳
StreamAllocation streamAllocation = new StreamAllocation(client.connectionPool(),
createAddress(request.url()), call, eventListener, callStackTrace);
this.streamAllocation = streamAllocation;
int followUpCount = 0;
Response priorResponse = null;
while (true) {
if (canceled) {
streamAllocation.release();
throw new IOException("Canceled");
}
Response response;
boolean releaseConnection = true;
try {
//將新建的流空間streamAllocation傳給下一個攔截器
response = realChain.proceed(request, streamAllocation, null, null);
releaseConnection = false;
} catch (RouteException e) {
// 路由異常后撩扒，判斷是否可以恢復(fù)
if (!recover(e.getLastConnectException(), streamAllocation, false, request)) {
throw e.getLastConnectException();
}
releaseConnection = false;
continue;
} catch (IOException e) {
// An attempt to communicate with a server failed. The request may have been sent.
boolean requestSendStarted = !(e instanceof ConnectionShutdownException);
if (!recover(e, streamAllocation, requestSendStarted, request)) throw e;
releaseConnection = false;
continue;
} finally {
// We're throwing an unchecked exception. Release any resources.
if (releaseConnection) {
streamAllocation.streamFailed(null);
streamAllocation.release();
}
}
//根據(jù)response的code修正request
Request followUp = followUpRequest(response, streamAllocation.route());
if (followUp == null) {
if (!forWebSocket) {
streamAllocation.release();
}
return response;
}
closeQuietly(response.body());
//重試次數(shù)超過最大次數(shù)后就放棄
if (++followUpCount > MAX_FOLLOW_UPS) {
streamAllocation.release();
throw new ProtocolException("Too many follow-up requests: " + followUpCount);
}
request = followUp;
priorResponse = response;
}
}
private boolean recover(IOException e, StreamAllocation streamAllocation,
boolean requestSendStarted, Request userRequest) {
streamAllocation.streamFailed(e);
// 是否運行重試
if (!client.retryOnConnectionFailure()) return false;
if (requestSendStarted && userRequest.body() instanceof UnrepeatableRequestBody) return false;
// 致命的異常不可恢復(fù)
if (!isRecoverable(e, requestSendStarted)) return false;
// No more routes to attempt.
if (!streamAllocation.hasMoreRoutes()) return false;
// For failure recovery, use the same route selector with a new connection.
return true;
}
}
重試攔截器的作用是當請求發(fā)生異常后，會根據(jù)異常進行修正和重試吨些。
整個攔截流程：RealInterceptorChain保存有所有的攔截器搓谆，通過它來分發(fā)調(diào)用攔截器，每個攔截器處理完自己的事情后豪墅，通過RealInterceptorChain再調(diào)用下一個攔截器泉手，直到獲得response
橋接攔截器 BridgeInterceptor
public final class BridgeInterceptor implements Interceptor {
public Response intercept(Chain chain) throws IOException {
Request userRequest = chain.request();
Request.Builder requestBuilder = userRequest.newBuilder();
RequestBody body = userRequest.body();
//補充頭信息
if (body != null) {
MediaType contentType = body.contentType();
if (contentType != null) {
requestBuilder.header("Content-Type", contentType.toString());
}
long contentLength = body.contentLength();
if (contentLength != -1) {
//http1.1的字段，同一個連接上面偶器，客戶端可以同時發(fā)送請求斩萌，通過該字段區(qū)分不同的請求
requestBuilder.header("Content-Length", Long.toString(contentLength));
requestBuilder.removeHeader("Transfer-Encoding");
} else {
//分塊傳輸，邊產(chǎn)生數(shù)據(jù)屏轰，邊返回颊郎，可提高效率
requestBuilder.header("Transfer-Encoding", "chunked");
requestBuilder.removeHeader("Content-Length");
}
}
if (userRequest.header("Host") == null) {
//http1.1的字段，支撐虛擬主機
requestBuilder.header("Host", hostHeader(userRequest.url(), false));
}
//連接復(fù)用霎苗，返回響應(yīng)后姆吭，不關(guān)閉連接
if (userRequest.header("Connection") == null) {
requestBuilder.header("Connection", "Keep-Alive");
}
// 頭部信息壓縮
boolean transparentGzip = false;
if (userRequest.header("Accept-Encoding") == null && userRequest.header("Range") == null) {
transparentGzip = true;
requestBuilder.header("Accept-Encoding", "gzip");
}
List<Cookie> cookies = cookieJar.loadForRequest(userRequest.url());
if (!cookies.isEmpty()) {
requestBuilder.header("Cookie", cookieHeader(cookies));
}
if (userRequest.header("User-Agent") == null) {
requestBuilder.header("User-Agent", Version.userAgent());
}
Response networkResponse = chain.proceed(requestBuilder.build());
HttpHeaders.receiveHeaders(cookieJar, userRequest.url(), networkResponse.headers());
Response.Builder responseBuilder = networkResponse.newBuilder()
.request(userRequest);
if (transparentGzip
&& "gzip".equalsIgnoreCase(networkResponse.header("Content-Encoding"))
&& HttpHeaders.hasBody(networkResponse)) {
GzipSource responseBody = new GzipSource(networkResponse.body().source());
Headers strippedHeaders = networkResponse.headers().newBuilder()
.removeAll("Content-Encoding")
.removeAll("Content-Length")
.build();
responseBuilder.headers(strippedHeaders);
String contentType = networkResponse.header("Content-Type");
responseBuilder.body(new RealResponseBody(contentType, -1L, Okio.buffer(responseBody)));
}
return responseBuilder.build();
}
}
連接攔截器
public final class ConnectInterceptor implements Interceptor {
public Response intercept(Chain chain) throws IOException {
RealInterceptorChain realChain = (RealInterceptorChain) chain;
Request request = realChain.request();
//這里的流空間是在RetryAndFollowUpInterceptor中創(chuàng)建的
StreamAllocation streamAllocation = realChain.streamAllocation();
// We need the network to satisfy this request. Possibly for validating a conditional GET.
boolean doExtensiveHealthChecks = !request.method().equals("GET");
HttpCodec httpCodec = streamAllocation.newStream(client, chain, doExtensiveHealthChecks);
RealConnection connection = streamAllocation.connection();
return realChain.proceed(request, streamAllocation, httpCodec, connection);
}
}
看一下streamAllocation.newStream()方法
public final class StreamAllocation {
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);
//新建http編碼器
HttpCodec resultCodec = resultConnection.newCodec(client, chain, this);
synchronized (connectionPool) {
codec = resultCodec;
return resultCodec;
}
} catch (IOException e) {
throw new RouteException(e);
}
}
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);
synchronized (connectionPool) {
if (candidate.successCount == 0) {
return candidate;
}
}
if (!candidate.isHealthy(doExtensiveHealthChecks)) {
noNewStreams();
continue;
}
return candidate;
}
}
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) {
//http1.1中允許連接復(fù)用，而且發(fā)請求不需要等上一個請求返回叨粘，但它是順序的猾编，一個連接只存在一個流空間
//http2中允許多路復(fù)用，一個連接同時發(fā)幾個請求升敲，也就是一個連接有多個流空間答倡，但流空間數(shù)量是有限制的，限制allocationLimit個
releasedConnection = this.connection;
//如果這個連接上面沒有流空間驴党，就釋放它
toClose = releaseIfNoNewStreams();
if (this.connection != null) {
result = this.connection;
releasedConnection = null;
}
if (!reportedAcquired) {
releasedConnection = null;
}
//當前連接不可用
if (result == null) {
//重新找可用的連接瘪撇，如果找到后，就新建一個流空間港庄，具體可以查看connectionPool里面的get方法
Internal.instance.get(connectionPool, address, this, null);
if (connection != null) {
//找到了可用的連接
foundPooledConnection = true;
result = connection;
} else {
selectedRoute = route;
}
}
}
//關(guān)閉socket
closeQuietly(toClose);
if (releasedConnection != null) {
eventListener.connectionReleased(call, releasedConnection);
}
if (foundPooledConnection) {
//通知從連接池里面找到了可用的連接
eventListener.connectionAcquired(call, result);
}
if (result != null) {
//找到可以用的連接倔既，返回
return result;
}
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) {
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;
}
}
}
//如果沒有找到，再創(chuàng)建一個新的連接
if (!foundPooledConnection) {
if (selectedRoute == null) {
selectedRoute = routeSelection.next();
}
//首次會執(zhí)行到這里鹏氧，創(chuàng)建連接
route = selectedRoute;
refusedStreamCount = 0;
result = new RealConnection(connectionPool, selectedRoute);
acquire(result, false);
}
}
if (foundPooledConnection) {
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;
//將連接放到連接池里面存起來
Internal.instance.put(connectionPool, result);
if (result.isMultiplexed()) {
socket = Internal.instance.deduplicate(connectionPool, address, this);
result = connection;
}
}
closeQuietly(socket);
eventListener.connectionAcquired(call, result);
return result;
}
//連接沒有流空間就釋放它
private Socket releaseIfNoNewStreams() {
assert (Thread.holdsLock(connectionPool));
RealConnection allocatedConnection = this.connection;
//allocatedConnection.noNewStreams之前介紹過渤涌，在connectPool中會定期清理空閑的連接，如果空閑把还，會通過noNewStreams標記
if (allocatedConnection != null && allocatedConnection.noNewStreams) {
return deallocate(false, false, true);
}
return null;
}
private Socket deallocate(boolean noNewStreams, boolean released, boolean streamFinished) {
assert (Thread.holdsLock(connectionPool));
if (streamFinished) {
this.codec = null;
}
if (released) {
this.released = true;
}
Socket socket = null;
if (connection != null) {
if (noNewStreams) {
connection.noNewStreams = true;
}
if (this.codec == null && (this.released || connection.noNewStreams)) {
//將連接里面的流空間釋放掉
release(connection);
if (connection.allocations.isEmpty()) {
connection.idleAtNanos = System.nanoTime();
//告訴connectionPool這個連接是閑置的实蓬，并用隊列里面移除
if (Internal.instance.connectionBecameIdle(connectionPool, connection)) {
//返回該連接的socket茸俭，需要關(guān)閉
socket = connection.socket();
}
}
connection = null;
}
}
return socket;
}
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));
}
}
總結(jié)一下：先判斷當前流空間里面的連接是否可用，可用就返回它安皱，不可用就釋放它调鬓，然后去連接池里面找，找不到就新建一個
看一下RealConnection
public final class RealConnection extends Http2Connection.Listener implements Connection {
public void connect(int connectTimeout, int readTimeout, int writeTimeout,
int pingIntervalMillis, boolean connectionRetryEnabled, Call call,
EventListener eventListener) {
while (true) {
try {
//是否是通過http來實現(xiàn)https
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);
}
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;
}
}
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();
}
}
}
//連接socket
private void connectSocket(int connectTimeout, int readTimeout, Call call,
EventListener eventListener) throws IOException {
Proxy proxy = route.proxy();
Address address = route.address();
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 {
//開始正在的連接
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;
}
// source是對socket的inputstream進行封裝酌伊，sink是對outputstream的封裝
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);
}
}
}
private void establishProtocol(ConnectionSpecSelector connectionSpecSelector,
int pingIntervalMillis, Call call, EventListener eventListener) throws IOException {
//http1.1的協(xié)議
if (route.address().sslSocketFactory() == null) {
protocol = Protocol.HTTP_1_1;
socket = rawSocket;
return;
}
eventListener.secureConnectStart(call);
connectTls(connectionSpecSelector);
eventListener.secureConnectEnd(call, handshake);
//http2的協(xié)議
if (protocol == Protocol.HTTP_2) {
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();
}
}
//找到連接后腾窝，會根據(jù)http1.1還是http2來生成對應(yīng)的解碼器
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);
}
}
}
現(xiàn)在再看一下CallServerInterceptor
public final class CallServerInterceptor implements Interceptor {
RealInterceptorChain realChain = (RealInterceptorChain) chain;
//我們先看http1.1的，Http1Codec
HttpCodec httpCodec = realChain.httpStream();
StreamAllocation streamAllocation = realChain.streamAllocation();
RealConnection connection = (RealConnection) realChain.connection();
Request request = realChain.request();
long sentRequestMillis = System.currentTimeMillis();
realChain.eventListener().requestHeadersStart(realChain.call());
//這里開始在outputstream里面寫數(shù)據(jù)居砖，開始發(fā)請求
httpCodec.writeRequestHeaders(request);
realChain.eventListener().requestHeadersEnd(realChain.call(), request);
Response.Builder responseBuilder = null;
//如果請求的body里面有數(shù)據(jù)
if (HttpMethod.permitsRequestBody(request.method()) && request.body() != null) {
// 100-continue表示先問服務(wù)器虹脯，是否接受我的大數(shù)據(jù)請求，如果接受悯蝉，我再將body發(fā)給你归形，節(jié)省帶寬
if ("100-continue".equalsIgnoreCase(request.header("Expect"))) {
httpCodec.flushRequest();
realChain.eventListener().responseHeadersStart(realChain.call());
responseBuilder = httpCodec.readResponseHeaders(true);
}
//為null表示可以接收數(shù)據(jù)
if (responseBuilder == null) {
// Write the request body if the "Expect: 100-continue" expectation was met.
realChain.eventListener().requestBodyStart(realChain.call());
long contentLength = request.body().contentLength();
//得到輸出流
CountingSink requestBodyOut =
new CountingSink(httpCodec.createRequestBody(request, contentLength));
//輸出流封裝緩存起來
BufferedSink bufferedRequestBody = Okio.buffer(requestBodyOut);
//將body的內(nèi)容輸出到輸出流里面
request.body().writeTo(bufferedRequestBody);
bufferedRequestBody.close();
realChain.eventListener()
.requestBodyEnd(realChain.call(), requestBodyOut.successfulCount);
} else if (!connection.isMultiplexed()) {
// If the "Expect: 100-continue" expectation wasn't met, prevent the HTTP/1 connection
// from being reused. Otherwise we're still obligated to transmit the request body to
// leave the connection in a consistent state.
streamAllocation.noNewStreams();
}
}
//將緩沖區(qū)的內(nèi)容發(fā)送出去
httpCodec.finishRequest();
if (responseBuilder == null) {
realChain.eventListener().responseHeadersStart(realChain.call());
//讀響應(yīng)的信息
responseBuilder = httpCodec.readResponseHeaders(false);
}
//封裝響應(yīng)的信息
Response response = responseBuilder
.request(request)
.handshake(streamAllocation.connection().handshake())
.sentRequestAtMillis(sentRequestMillis)
.receivedResponseAtMillis(System.currentTimeMillis())
.build();
int code = response.code();
return response;
}
看一下Http1Codec的實現(xiàn)
public final class Http1Codec implements HttpCodec {
//發(fā)送請求
public void writeRequestHeaders(Request request) throws IOException {
//requestLine里面包含url信息
String requestLine = RequestLine.get(
request, streamAllocation.connection().route().proxy().type());
writeRequest(request.headers(), requestLine);
}
//直接將請求發(fā)給服務(wù)器
public void writeRequest(Headers headers, String requestLine) throws IOException {
if (state != STATE_IDLE) throw new IllegalStateException("state: " + state);
sink.writeUtf8(requestLine).writeUtf8("\r\n");
for (int i = 0, size = headers.size(); i < size; i++) {
sink.writeUtf8(headers.name(i))
.writeUtf8(": ")
.writeUtf8(headers.value(i))
.writeUtf8("\r\n");
}
sink.writeUtf8("\r\n");
state = STATE_OPEN_REQUEST_BODY;
}
//將緩沖區(qū)的數(shù)據(jù)發(fā)送出去
public void flushRequest() throws IOException {
sink.flush();
}
public void finishRequest() throws IOException {
sink.flush();
}
//讀取響應(yīng)的信息
public Response.Builder readResponseHeaders(boolean expectContinue) throws IOException {
try {
StatusLine statusLine = StatusLine.parse(readHeaderLine());
Response.Builder responseBuilder = new Response.Builder()
.protocol(statusLine.protocol)
.code(statusLine.code)
.message(statusLine.message)
.headers(readHeaders());
if (expectContinue && statusLine.code == HTTP_CONTINUE) {
return null;
} else if (statusLine.code == HTTP_CONTINUE) {
state = STATE_READ_RESPONSE_HEADERS;
return responseBuilder;
}
state = STATE_OPEN_RESPONSE_BODY;
return responseBuilder;
} catch (EOFException e) {
// Provide more context if the server ends the stream before sending a response.
IOException exception = new IOException("unexpected end of stream on " + streamAllocation);
exception.initCause(e);
throw exception;
}
}
//讀取響應(yīng)信息的原始數(shù)據(jù)
private String readHeaderLine() throws IOException {
String line = source.readUtf8LineStrict(headerLimit);
headerLimit -= line.length();
return line;
}
}
好了，發(fā)送請求和讀取響應(yīng)數(shù)據(jù)就解析完了
