Android开发okhttp源码分析

释放双眼,带上耳机,听听看~!

1.OkHttp同步的使用

//如果创建创建需要很多的参数,可以使用Builder()方式
OkHttpClient mOkHttpClient = new OkHttpClient.Builder().readTimeout(5, TimeUnit.SECONDS).build();
public void synRequest(){
Request  request=new Request.Builder().url("http://www.baidu.com").get().build();
Call call=mOkHttpClient.newCall(request);
try{
Response response=call.execute();
System.out.println(response.body().toString());
}catch(Exception e){
e.printStackTrace();
}
}

1看下OkHttpClient的Builder()方法,对象属性的初始化

 public Builder() {
dispatcher = new Dispatcher(); //请求分发器,由它决定异步请求时直接处理还是缓存等待,对于同步请求没有做太多的操作
protocols = DEFAULT_PROTOCOLS;
connectionSpecs = DEFAULT_CONNECTION_SPECS;
eventListenerFactory = EventListener.factory(EventListener.NONE);
proxySelector = ProxySelector.getDefault();
if (proxySelector == null) {
proxySelector = new NullProxySelector();
}
cookieJar = CookieJar.NO_COOKIES;
socketFactory = SocketFactory.getDefault();
hostnameVerifier = OkHostnameVerifier.INSTANCE;
certificatePinner = CertificatePinner.DEFAULT;
proxyAuthenticator = Authenticator.NONE;
authenticator = Authenticator.NONE;
connectionPool = new ConnectionPool();//它是一个连接池,客户端和服务器之间的连接可以抽象为一个connection,每一个connection都会放在连接池中,由它来进行统一的管理,要是相同的话,可以选择复用,它可以决哪些连接是可以保持打开,哪些是可以复用的

2.看下Request请求报文的Builder()方法

public Builder() {
this.method = "GET";
this.headers = new Headers.Builder();
}

3.看下Request的build()方法

public Request build() {
if (url == null) throw new IllegalStateException("url == null");
return new Request(this);
}

4.根据3看下request的构造方法,方法里是有关请求信息的属性

Request(Builder builder) {
this.url = builder.url;
this.method = builder.method;
this.headers = builder.headers.build();
this.body = builder.body;
this.tags = Util.immutableMap(builder.tags);
}

5.看下newCall(request)方法一直跟踪到call的实现类RealCall构造方法,可以看到RealCall持有了前两步创建的OkHttpClient和Request

private RealCall(OkHttpClient client, Request originalRequest, boolean forWebSocket) {
this.client = client;
this.originalRequest = originalRequest;
this.forWebSocket = forWebSocket;
}

6.接下来是call.execute()

 @Override public Response execute() throws IOException {
synchronized (this) {
if (executed) throw new IllegalStateException("Already Executed"); //执行过就报异常
executed = true
}
transmitter.timeoutEnter();
transmitter.callStart();
try {
client.dispatcher().executed(this);
return getResponseWithInterceptorChain(); //拦截器链的操作
} finally {
client.dispatcher().finished(this);//主动回收同步请求
}
}

7.看下client.dispatcher()就返回dispatcher实例
public Dispatcher dispatcher() {
return dispatcher;
}
8.看下 client.dispatcher().executed(this); 就是在分发器里同步请求队列了添加请求

  /** Used by {@code Call#execute} to signal it is in-flight. */
synchronized void executed(RealCall call) {
runningSyncCalls.add(call);
}
/** Running synchronous calls. Includes canceled calls that haven't finished yet. */
private final Deque<RealCall> runningSyncCalls = new ArrayDeque<>();

9.看下finished()方法

private <T> void finished(Deque<T> calls, T call) {
Runnable idleCallback;
synchronized (this) {
if (!calls.remove(call)) throw new AssertionError("Call wasn't in-flight!"); //不能移除同步请求报异常
idleCallback = this.idleCallback;
}
boolean isRunning = promoteAndExecute();
if (!isRunning && idleCallback != null) {
idleCallback.run();
}

注意 发送同步请求后,就会进入阻塞状态,直到收到响应

3.OkHttp异步的使用*

 OkHttpClient mOkHttpClient = new OkHttpClient.Builder().readTimeout(5, TimeUnit.SECONDS).build();
public void synRequest(){
Request  request=new Request.Builder().url("http://www.baidu.com").get().build();
Call call=mOkHttpClient.newCall(request);
call.enqueue(new Callback(){//前三步都没有发起网络请求,这步才开始
@Override
public void onFailure(Call call, IOException e) {
}
@Override
public void onResponse(Call call, Response response) throws IOException {
System.out.println(response.body().toString());
}
});
}
1.看下enqueue()方法
@Override public void enqueue(Callback responseCallback) {
synchronized (this) {
if (executed) throw new IllegalStateException("Already Executed");
executed = true;//call只能被执行一次
}
transmitter.callStart();
client.dispatcher().enqueue(new AsyncCall(responseCallback));
}
2.看下client.dispatcher().enqueue(new AsyncCall(responseCallback));enqueue()方法
void enqueue(AsyncCall call) {
synchronized (this) {
readyAsyncCalls.add(call);//缓存等待的队列添加异步请求
// Mutate the AsyncCall so that it shares the AtomicInteger of an existing running call to
// the same host.
if (!call.get().forWebSocket) {
AsyncCall existingCall = findExistingCallWithHost(call.host());
if (existingCall != null) call.reuseCallsPerHostFrom(existingCall);
}
}
promoteAndExecute();
}
/** Ready async calls in the order they'll be run. */
private final Deque<AsyncCall> readyAsyncCalls = new ArrayDeque<>(); //缓存等待的队列
/** Running asynchronous calls. Includes canceled calls that haven't finished yet. */
private final Deque<AsyncCall> runningAsyncCalls = new ArrayDeque<>();  //异步队列

3.看下promoteAndExecute() 方法

 private boolean promoteAndExecute() {
assert (!Thread.holdsLock(this));
List<AsyncCall> executableCalls = new ArrayList<>();
boolean isRunning;
synchronized (this) {
for (Iterator<AsyncCall> i = readyAsyncCalls.iterator(); i.hasNext(); ) {
AsyncCall asyncCall = i.next();
if (runningAsyncCalls.size() >= maxRequests) break; // 异步请求队列里请求数量要小于64.
if (asyncCall.callsPerHost().get() >= maxRequestsPerHost) continue; // Host max capacity.
i.remove();//遍历缓存队列,移除,添加到正在运行的队列
asyncCall.callsPerHost().incrementAndGet();
executableCalls.add(asyncCall);
runningAsyncCalls.add(asyncCall);//小于64就添加到异步请求对ie
}
isRunning = runningCallsCount() > 0;
}
for (int i = 0, size = executableCalls.size(); i < size; i++) {
AsyncCall asyncCall = executableCalls.get(i);
asyncCall.executeOn(executorService());//执行
}
return isRunning;
}
private int maxRequests = 64;
private int maxRequestsPerHost = 5;
4.看下runningCallsCount()方法
public synchronized int runningCallsCount() {
return runningAsyncCalls.size() + runningSyncCalls.size();//同步请求队列里请求的数量和异步请求队列里请求的数量
}

5.看下 asyncCall.executeOn(executorService());里的executorService

 public synchronized ExecutorService executorService() { //同步
if (executorService == null) { //保证单例
executorService = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS,  //线程池 Integer.MAX_VALUE前面有maxRequests=64限制 参数的意义,60s后,将无关的线程关闭
new SynchronousQueue<>(), Util.threadFactory("OkHttp Dispatcher", false));
}
return executorService;
}

6.看下AsyncCall,AsyncCall就是Runnable

 final class AsyncCall extends NamedRunnable {
public abstract class NamedRunnable implements Runnable {

7.看下asyncCall.executeOn()方法

    @Override protected void execute() {
boolean signalledCallback = false;
transmitter.timeoutEnter();
try {
Response response = getResponseWithInterceptorChain();//拦截器链
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 {
responseCallback.onFailure(RealCall.this, e);//子线程
}
} finally {
client.dispatcher().finished(this);//1.从正在执行的异步队列中,移除当前请求 2.重新计算正在运行请求的数量
}
}
}

一个异步请求执行完后,要将此请求从异步执行队列中移除
注意onFailure,onResponse方法都是在子线程中进行的

dispatcher
dispatcher的作用为维护请求的状态,并维护一个线程池,用于执行请求
dispatcher里为什么要维护两个异步请求队列
dispatcher运行在主线程的生产者
ExecutorService消费者池
Deque readyAsyncCalls缓存异步队列
Deque readyAsyncCalls正在执行的异步队列

拦截器是okhttp中提供一种强大机制,它可以实现网监听,请求以及响应重写,请求失败重功能(不区分同步异步)

Response getResponseWithInterceptorChain() throws IOException {
// Build a full stack of interceptors.
List<Interceptor> interceptors = new ArrayList<>();
interceptors.addAll(client.interceptors());
interceptors.add(new RetryAndFollowUpInterceptor(client));
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));
Interceptor.Chain chain = new RealInterceptorChain(interceptors, transmitter, null, 0,
originalRequest, this, client.connectTimeoutMillis(),
client.readTimeoutMillis(), client.writeTimeoutMillis());
boolean calledNoMoreExchanges = false;
try {
Response response = chain.proceed(originalRequest);
1.看下 chain.proceed(originalRequest)方法
public Response proceed(Request request, Transmitter transmitter, @Nullable Exchange exchange)
throws IOException {
if (index >= interceptors.size()) throw new AssertionError();
calls++;
// If we already have a stream, confirm that the incoming request will use it.
if (this.exchange != null && !this.exchange.connection().supportsUrl(request.url())) {
throw new IllegalStateException("network interceptor " + interceptors.get(index - 1)
+ " must retain the same host and port");
}
// If we already have a stream, confirm that this is the only call to chain.proceed().
if (this.exchange != null && calls > 1) {
throw new IllegalStateException("network interceptor " + interceptors.get(index - 1)
+ " must call proceed() exactly once");
}
// Call the next interceptor in the chain.
RealInterceptorChain next = new RealInterceptorChain(interceptors, transmitter, exchange,
index + 1, request, call, connectTimeout, readTimeout, writeTimeout); //从index + 1开始访问,也就是下一个拦截器开始访问,构成拦截器链
Interceptor interceptor = interceptors.get(index);
Response response = interceptor.intercept(next);

1.RetryAndFollowUpInterceptor拦截器 此拦截器根据需要执行重定向
看RetryAndFollowUpInterceptor的intercept方法

 @Override public Response intercept(Chain chain) throws IOException {
.
.
.
if (++followUpCount > MAX_FOLLOW_UPS) {
throw new ProtocolException("Too many follow-up requests: " + followUpCount);
}
request = followUp;
priorResponse = response;
}
private static final int MAX_FOLLOW_UPS = 20; //重连次数20次

2.BridgeInterceptor拦截器
从应用程序代码到网络代码的桥梁。首先,它构建来自用户的网络请求
*请求。然后它继续呼叫网络。最后,它从网络构建用户响应回应。

  • 看BridgeInterceptor的intercept()方法
 *  `@Override public Response intercept(Chain chain) throws IOException {`
.
.
.
if (userRequest.header("Connection") == null) {
requestBuilder.header("Connection", "Keep-Alive");//保持连接
.
.
.
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")) //支持gzip
&& 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)));
}
}
}

3.okhttp的缓存
使用

  OkHttpClient mOkHttpClient = new OkHttpClient.Builder().cache(new Cache(new File("catch"),24*1024*1024)).readTimeout(5, TimeUnit.SECONDS).build();

看下cache的put方法

 @Nullable CacheRequest put(Response response) {
String requestMethod = response.request().method();
if (HttpMethod.invalidatesCache(response.request().method())) {
try {
remove(response.request());
} catch (IOException ignored) {
// The cache cannot be written.
}
return null;
}
if (!requestMethod.equals("GET")) {  //不缓存get方法
// Don't cache non-GET responses. We're technically allowed to cache
// HEAD requests and some POST requests, but the complexity of doing
// so is high and the benefit is low.
return null;
}
if (HttpHeaders.hasVaryAll(response)) {
return null;
}
Entry entry = new Entry(response); //它的实例就是要写入缓存的部分(它的源码里都是有关需要缓存的属性,被封装成了Entry)
DiskLruCache.Editor editor = null; //okhttp的缓存技术就是使用DiskLruCache
try {
editor = cache.edit(key(response.request().url()));//将网络请求的url转换为对应的key(源码里是MD5加密的十六进制的表现形式)
if (editor == null) {
return null;
}
entry.writeTo(editor);//开始缓存
return new CacheRequestImpl(editor);
} catch (IOException e) {
abortQuietly(editor);
return null;
}
}
看下writeTo()方法
```java
public void writeTo(DiskLruCache.Editor editor) throws IOException {
BufferedSink sink = Okio.buffer(editor.newSink(ENTRY_METADATA));
.
.
.
```if (isHttps()) {  //判断了是否是https请求,写入了握手的信息
sink.writeByte('\n');
sink.writeUtf8(handshake.cipherSuite().javaName())
.writeByte('\n');
writeCertList(sink, handshake.peerCertificates());
writeCertList(sink, handshake.localCertificates());
sink.writeUtf8(handshake.tlsVersion().javaName()).writeByte('\n');
}
sink.close();
}

还有主体信息在哪缓存?看下CacheRequestImpl(final DiskLruCache.Editor editor)方法

CacheRequestImpl(final DiskLruCache.Editor editor) { //这个方法写入body信息
this.editor = editor;
this.cacheOut = editor.newSink(ENTRY_BODY);
this.body = new ForwardingSink(cacheOut) {
@Override public void close() throws IOException {
synchronized (Cache.this) {
if (done) {
return;
}
done = true;
writeSuccessCount++;
}
super.close();
editor.commit();
}
};
}

CacheInterceptor缓存拦截器
看下intercept方法

  @Override public Response intercept(Chain chain) throws IOException {
Response cacheCandidate = cache != null
? cache.get(chain.request()) //获取缓存
: null;
long now = System.currentTimeMillis();
CacheStrategy strategy = new CacheStrategy.Factory(now, chain.request(), cacheCandidate).get();//工厂方法,实例化CacheStrategy ,写入请求头的操作   
/*
/** CacheStrategy 缓存策略,看是从networkRequest网络请求还是cacheResponse缓存中获取Response;还可以同时使用,进行对比,选择最好的策略
The request to send on the network, or null if this call doesn't use the network. */
public final @Nullable Request networkRequest;
/** The cached response to return or validate; or null if this call doesn't use a cache. */
public final @Nullable Response cacheResponse;
**/
Request networkRequest = strategy.networkRequest;
Response cacheResponse = strategy.cacheResponse;
if (cache != null) {
cache.trackResponse(strategy);//更新缓存指标,主要是命中率,在okhttp3.Cache的trackResponse()方法里,hitCount++表示命中了缓存
}
if (cacheCandidate != null && cacheResponse == null) {
closeQuietly(cacheCandidate.body()); // The cache candidate wasn't applicable. Close it.
}
// If we're forbidden from using the network and the cache is insufficient, fail.
if (networkRequest == null && cacheResponse == null) {  //网络不好,没有缓存,构建Response,抛出504错误
return new Response.Builder()
.request(chain.request())
.protocol(Protocol.HTTP_1_1)
.code(504)
.message("Unsatisfiable Request (only-if-cached)")
.body(Util.EMPTY_RESPONSE)
.sentRequestAtMillis(-1L)
.receivedResponseAtMillis(System.currentTimeMillis())
.build();
}
// If we don't need the network, we're done.
if (networkRequest == null) {  //不需要网络,有缓存,直接返回缓存
return cacheResponse.newBuilder()
.cacheResponse(stripBody(cacheResponse))
.build();
}
Response networkResponse = null;
try {
networkResponse = chain.proceed(networkRequest);//交给下一个拦截器工作
} finally {
// If we're crashing on I/O or otherwise, don't leak the cache body.
if (networkResponse == null && cacheCandidate != null) {
closeQuietly(cacheCandidate.body());
}
}
// If we have a cache response too, then we're doing a conditional get.
if (cacheResponse != null) {
if (networkResponse.code() == HTTP_NOT_MODIFIED) { //如果响应码HTTP_NOT_MODIFIED为304,直接从缓存中读取数据
Response response = cacheResponse.newBuilder()
.headers(combine(cacheResponse.headers(), networkResponse.headers()))
.sentRequestAtMillis(networkResponse.sentRequestAtMillis())
.receivedResponseAtMillis(networkResponse.receivedResponseAtMillis())
.cacheResponse(stripBody(cacheResponse))
.networkResponse(stripBody(networkResponse))
.build();
networkResponse.body().close();
// Update the cache after combining headers but before stripping the
// Content-Encoding header (as performed by initContentStream()).
cache.trackConditionalCacheHit();
cache.update(cacheResponse, response);
return response;
} else {
closeQuietly(cacheResponse.body());
}
}
Response response = networkResponse.newBuilder()
.cacheResponse(stripBody(cacheResponse))
.networkResponse(stripBody(networkResponse))
.build();
if (cache != null) {
if (HttpHeaders.hasBody(response) && CacheStrategy.isCacheable(response, networkRequest)) {//如果有响应体,缓存策略可以缓存,直接缓存
// Offer this request to the cache.
CacheRequest cacheRequest = cache.put(response);
return cacheWritingResponse(cacheRequest, response);
}
if (HttpMethod.invalidatesCache(networkRequest.method())) {//无效的缓存方法,移除request        
try {
cache.remove(networkRequest);
} catch (IOException ignored) {
// The cache cannot be written.
}
}
}
return response;
}

看下CacheStrategy 的getCandidate()方法

private CacheStrategy getCandidate() {
.
.
.
//写入请求头的操作                      
if (!responseCaching.noCache() && ageMillis + minFreshMillis < freshMillis + maxStaleMillis) {
Response.Builder builder = cacheResponse.newBuilder();
if (ageMillis + minFreshMillis >= freshMillis) {
builder.addHeader("Warning", "110 HttpURLConnection \"Response is stale\"");
}
long oneDayMillis = 24 * 60 * 60 * 1000L;
if (ageMillis > oneDayMillis && isFreshnessLifetimeHeuristic()) {
builder.addHeader("Warning", "113 HttpURLConnection \"Heuristic expiration\"");
}
return new CacheStrategy(null, builder.build());
}
String conditionName;
String conditionValue;
if (etag != null) {
conditionName = "If-None-Match";
conditionValue = etag;
} else if (lastModified != null) {
conditionName = "If-Modified-Since";
conditionValue = lastModifiedString;
} else if (servedDate != null) {
conditionName = "If-Modified-Since";
conditionValue = servedDateString;
} else {
return new CacheStrategy(request, null); // No condition! Make a regular request.
}
Headers.Builder conditionalRequestHeaders = request.headers().newBuilder();
Internal.instance.addLenient(conditionalRequestHeaders, conditionName, conditionValue);
Request conditionalRequest = request.newBuilder()
.headers(conditionalRequestHeaders.build())
.build();
return new CacheStrategy(conditionalRequest, cacheResponse);
}

ConnectInterceptor网络连接拦截器
看下intercept方法

@Override public Response intercept(Chain chain) throws IOException {
.
.
.
Exchange exchange = transmitter.newExchange(chain, doExtensiveHealthChecks);
return realChain.proceed(request, transmitter, exchange);
}

看下newExchange()方法

  Exchange newExchange(Interceptor.Chain chain, boolean doExtensiveHealthChecks) {
.
.
.
ExchangeCodec codec = exchangeFinder.find(client, chain, doExtensiveHealthChecks);
}
``
看下 exchangeFinder.find(client, chain, doExtensiveHealthChecks)方法
```java
public ExchangeCodec find(
.
.
.
try {
RealConnection resultConnection = findHealthyConnection(connectTimeout, readTimeout,//RealConnection 是核心
writeTimeout, pingIntervalMillis, connectionRetryEnabled, doExtensiveHealthChecks);
return resultConnection.newCodec(client, chain);
} catch (RouteException e) {
trackFailure();
throw e;
} catch (IOException e) {
trackFailure();
throw new RouteException(e);
}
}
}

看下findHealthyConnection()方法

private RealConnection findHealthyConnection(int connectTimeout, int readTimeout,
.
.
.
synchronized (connectionPool) {
if (candidate.successCount == 0) {  //如果successCount =0,就返回,这时候循环已结束
return candidate;
}
if (!candidate.isHealthy(doExtensiveHealthChecks)) { //检查candidate是否健康(比如Socket连接没有关闭,输入输出流没有关闭,就是不健康的)
candidate.noNewExchanges(); //如果不健康,进行资源销毁的动作
continue;//循环调用RealConnection candidate = findConnection(connectTimeout, readTimeout, writeTimeout,
pingIntervalMillis, connectionRetryEnabled);去寻找RealConnection
}
return candidate;
}
}

看下 findConnection(connectTimeout, readTimeout, writeTimeout, pingIntervalMillis, connectionRetryEnabled);方法

 private RealConnection findConnection(int connectTimeout, int readTimeout, int writeTimeout,
int pingIntervalMillis, boolean connectionRetryEnabled) throws IOException {
.
.
.
if (transmitter.connection != null) {
result = transmitter.connection;  //尝试复用,赋值给result,result就是RealConnection
releasedConnection = null;
}
.
.
.
if (result == null) {
// 如果不能复用从connectionPool连接池中获取
if (connectionPool.transmitterAcquirePooledConnection(address, transmitter, null, false)) {
foundPooledConnection = true;
result = transmitter.connection;
} else if (nextRouteToTry != null) {
selectedRoute = nextRouteToTry;
nextRouteToTry = null;
} else if (retryCurrentRoute()) {
selectedRoute = transmitter.connection.route();
}
}
}
.
.
.
result.connect(connectTimeout, readTimeout, writeTimeout, pingIntervalMillis,//核心,获取到了RealConnection,就进行网络连接
connectionRetryEnabled, call, eventListener);
connectionPool.routeDatabase.connected(result.route());
Socket socket = null;
synchronized (connectionPool) {
connectingConnection = null;
// Last attempt at connection coalescing, which only occurs if we attempted multiple
// concurrent connections to the same host.
if (connectionPool.transmitterAcquirePooledConnection(address, transmitter, routes, true)) {
// We lost the race! Close the connection we created and return the pooled connection.
result.noNewExchanges = true;
socket = result.socket();
result = transmitter.connection;
} else {
connectionPool.put(result);//获取到了RealConnection,添加到连接池中
transmitter.acquireConnectionNoEvents(result);
}
}
}

看下esult.connect(connectTimeout, readTimeout, writeTimeout, pingIntervalMillis,connectionRetryEnabled, call, eventListener);方法

 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();//指定Socket连接的配置
ConnectionSpecSelector connectionSpecSelector = new ConnectionSpecSelector(connectionSpecs);//用于选择链接(隧道链接,Socket链接)
.
.
.
}

RealConnectionPool连接池 作用:在它的时间范围内复用connection
看下 transmitterAcquirePooledConnection(Address address, Transmitter transmitter,@Nullable List routes, boolean requireMultiplexed) 获取链接的方法

  boolean transmitterAcquirePooledConnection(Address address, Transmitter transmitter,
@Nullable List<Route> routes, boolean requireMultiplexed) {
assert (Thread.holdsLock(this));
for (RealConnection connection : connections) {
if (requireMultiplexed && !connection.isMultiplexed()) continue;
if (!connection.isEligible(address, routes)) continue;//判断链接是否可以使用
transmitter.acquireConnectionNoEvents(connection);
return true;
}
return false;
}

看下transmitter.acquireConnectionNoEvents(connection)方法

void acquireConnectionNoEvents(RealConnection connection) {
assert (Thread.holdsLock(connectionPool));
if (this.connection != null) throw new IllegalStateException();
this.connection = connection; //赋值
connection.transmitters.add(new TransmitterReference(this, callStackTrace));//添加到transmitters,transmitters就是一个集合,通过集合的大小判断是否超过负载量
}

看下put(RealConnection connection) 方法

  void put(RealConnection connection) {
assert (Thread.holdsLock(this));
if (!cleanupRunning) {
cleanupRunning = true;
executor.execute(cleanupRunnable);//执行
}
connections.add(connection);//添加到connections集合中
}

看下清理线程池的代码

  private final Runnable cleanupRunnable = () -> {
while (true) {
long waitNanos = cleanup(System.nanoTime());//下一次需要清理间隔的时间
if (waitNanos == -1) return;
if (waitNanos > 0) {
long waitMillis = waitNanos / 1000000L;
waitNanos -= (waitMillis * 1000000L);
synchronized (RealConnectionPool.this) {
try {
RealConnectionPool.this.wait(waitMillis, (int) waitNanos);//等待释放锁和时间片,当等待的时间过了之后,它会再次调用Runnable去清理,同时会返回下次需要清理的间隔时间
} catch (InterruptedException ignored) {
}
}
}
}
};

看下 cleanup(System.nanoTime())方法 相当于虚拟机的标记清除算法,就是标记不活跃或者长时间为使用的链接,然后进行清除

  long cleanup(long now) {
int inUseConnectionCount = 0;
int idleConnectionCount = 0;
RealConnection longestIdleConnection = null;
long longestIdleDurationNs = Long.MIN_VALUE;
// Find either a connection to evict, or the time that the next eviction is due.
synchronized (this) {
for (Iterator<RealConnection> i = connections.iterator(); i.hasNext(); ) {
RealConnection connection = i.next();
// If the connection is in use, keep searching.
if (pruneAndGetAllocationCount(connection, now) > 0) {//找到正在使用的链接的方法
inUseConnectionCount++;
continue;
}
idleConnectionCount++;
// If the connection is ready to be evicted, we're done.
long idleDurationNs = now - connection.idleAtNanos;
if (idleDurationNs > longestIdleDurationNs) {
longestIdleDurationNs = idleDurationNs;
longestIdleConnection = connection;
}
}
if (longestIdleDurationNs >= this.keepAliveDurationNs
|| idleConnectionCount > this.maxIdleConnections) {
// We've found a connection to evict. Remove it from the list, then close it below (outside
// of the synchronized block).
connections.remove(longestIdleConnection);//空闲链接数量超过数量,就进行清除
} else if (idleConnectionCount > 0) {
// A connection will be ready to evict soon.
return keepAliveDurationNs - longestIdleDurationNs;
} else if (inUseConnectionCount > 0) {
// All connections are in use. It'll be at least the keep alive duration 'til we run again.
return keepAliveDurationNs;
} else {
// No connections, idle or in use.
cleanupRunning = false;
return -1;//没有需要清理的链接,跳出循环
}
}
closeQuietly(longestIdleConnection.socket());
// Cleanup again immediately.
return 0;
}

CallServerInterceptor拦截器,主要是向服务器发起一个真正的网络请求,接收到服务器给我们的返回响应
看下@Override public Response intercept(Chain chain) throws IOException 方法

@Override public Response intercept(Chain chain) throws IOException {
RealInterceptorChain realChain = (RealInterceptorChain) chain;
Exchange exchange = realChain.exchange(); //编码request,解码response
Request request = realChain.request();
long sentRequestMillis = System.currentTimeMillis();
exchange.writeRequestHeaders(request);//向Socket里写入请求的头部信息
boolean responseHeadersStarted = false;
Response.Builder responseBuilder = null;
if ("100-continue".equalsIgnoreCase(request.header("Expect"))) { //特殊处理,头部是否可以包含Expect: 100-continue
exchange.flushRequest();
responseHeadersStarted = true;
exchange.responseHeadersStart();
responseBuilder = exchange.readResponseHeaders(true);
}
if (responseBuilder == null) {
if (request.body().isDuplex()) {
// Prepare a duplex body so that the application can send a request body later.
exchange.flushRequest();
BufferedSink bufferedRequestBody = Okio.buffer(
exchange.createRequestBody(request, true));
request.body().writeTo(bufferedRequestBody);//向Socket里写入body信息
.
.
.
if (request.body() == null || !request.body().isDuplex()) {
exchange.finishRequest();//完成整个网络请求的写入信息
}
if (responseBuilder == null) {
.
.
.
responseBuilder = exchange.readResponseHeaders(false);//读取网络请求的头部信息
.
.
.
if (forWebSocket && code == 101) { //判断响应码
// Connection is upgrading, but we need to ensure interceptors see a non-null response body.
response = response.newBuilder()
.body(Util.EMPTY_RESPONSE)
.build();
} else {
response = response.newBuilder()  //构建response
.body(exchange.openResponseBody(response))
.build();
}
if ("close".equalsIgnoreCase(response.request().header("Connection"))
|| "close".equalsIgnoreCase(response.header("Connection"))) {
exchange.noNewExchangesOnConnection();//关闭流,关闭链接
}

为TA充电
共{{data.count}}人
人已赞赏
Android文章

Android开发Glide源码解析

2021-2-1 23:01:07

Android文章

Android开发Retrofit源码解析

2021-2-1 23:55:01

0 条回复 A文章作者 M管理员
    暂无讨论,说说你的看法吧
个人中心
购物车
优惠劵
今日签到
有新私信 私信列表
搜索