一、OkHttp的基本流程
1. OkHttp的基本流程
OkHttpClient、Request -> RealCall -> Dispatcher -> interceptors -> RetryAndFollowInterceptor -> BirdgeInterceptor -> CacheInterceptor -> ConnectInterceptor -> CallServerInterceptor
2. 相关概念
2.1 OkHttpClient
OKHTTP使用OkHttpClient来发送和读取请求的响应,OkHttpClient在使用时,应该被设置为单例,只实例化一次,应该每个OkHttpClient都包含一个连接池和线程池。重用OkHttpClient可以减少延迟并节省内存
2.2 Request
OkHttp的请求
2.3 Dispatcher
Dispatch主要用于控制并发的请求,请求执行的策略
2.4 interceptors
拦截器是OkHttp中提供一种强大机制,它可以实现网络监听、请求以及响应重写,请求失败重试等功能
2.5 RetryAndFollowUpInterceptor
重试重定向拦截器
2.6 BridgeInterceptor
桥接拦截器,负责将用户构建的request转化为能够进行网络访问的请求,然后发起请求,最后将这个网络请求回来的响应Response转化为用户的Response。
2.7 CacheInterceptor
缓存拦截器,负责写入和读取缓存
2.8 ConnectInterceptor
链接拦截器,与服务器创建连接,并将其传给下一个拦截器
2.9 CallServerInterceptor
发起请求拦截器,这是最后一个拦截器,会对服务器进行调用
二、OkHttp源码解析
1 从请求处理开始分析
1.1 请求步骤
- 创建一个OkHttpClient对象
- 构建一个Request对象,通过OkHttpClient和Request对象,构建出Call对象
- 执行Call的execute(同步)/enqueue(异步)方法
1.2 请求源码分析
请求网络需要用OkHttpClient.newCall(request)进行execute或者enqueue操作:当调用newCall方法是,会调用如下代码:
/** * Prepares the {@code request} to be executed at some point in the future. */ @Override public Call newCall(Request request) { return new RealCall(this, request); }复制代码
其实际返回的是一个RealCall类。调用enqueue异步请求网络实际上是调用了RealCall的enqueue方法。查看RealCall的enqueue方法
@Override public void enqueue(Callback responseCallback) { synchronized (this) { if (executed) throw new IllegalStateException("Already Executed"); executed = true; } client.dispatcher().enqueue(new AsyncCall(responseCallback)); }复制代码
最终的请求是dispatcher来完成的。
2 Dispatcher任务调度
2.1 Dispatcher源码分析
Dispatch主要用于控制并发的请求,它主要维护了以下变量:
/** 最大并发请求数 */ private int maxRequests = 64; /** 最大主机请求数 */ private int maxRequestsPerHost = 5; private Runnable idleCallback; /** 消费者线程池 */ private ExecutorService executorService; /** 准备运行的异步请求队列 */ private final DequereadyAsyncCalls = new ArrayDeque<>(); /** 正在运行的异步请求队列 */ private final Deque runningAsyncCalls = new ArrayDeque<>(); /** 正在运行的同步请求队列 */ private final Deque runningSyncCalls = new ArrayDeque<>();复制代码
Dispatcher的构造方法:
public Dispatcher(ExecutorService executorService) { this.executorService = executorService; } public Dispatcher() { } public synchronized ExecutorService executorService() { if (executorService == null) { executorService = new ThreadPoolExecutor(0, Integer.MAX_VALUE, 60, TimeUnit.SECONDS, new SynchronousQueue(), Util.threadFactory("OkHttp Dispatcher", false)); } return executorService; }复制代码
默认线程池,核心线程数为0,最大线程数为整型最大值,空闲时间为60s。
当调用RealCall的enqueue方法时,实际上是调用Dispatch的enqueue方法:synchronized void enqueue(AsyncCall call) { if (runningAsyncCalls.size() < maxRequests && runningCallsForHost(call) < maxRequestsPerHost) { runningAsyncCalls.add(call); executorService().execute(call); } else { readyAsyncCalls.add(call); } }复制代码
当正在运行的异步请求队列中的数量小于64并且正在运行的请求主机数小于5时,把请求加载到runningAsynCalls中并在线程池中执行,否则就加入到readyAsyncCalls中进行缓存等待。线程池中传进来的参数是AsyncCall,它是RealCall内部类,其内部也实现了execute方法
@Override protected void execute() { boolean signalledCallback = false; try { 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 { responseCallback.onFailure(RealCall.this, e); } } finally { client.dispatcher().finished(this); } }复制代码
在上面代码的finally中,可以看出无论这个请求的结果如何,都会执行client.dispatcher().finished(this),finished方法如下:
privatevoid finished(Deque calls, T call, boolean promoteCalls) { int runningCallsCount; Runnable idleCallback; synchronized (this) { if (!calls.remove(call)) throw new AssertionError("Call wasn't in-flight!"); if (promoteCalls) promoteCalls(); runningCallsCount = runningCallsCount(); idleCallback = this.idleCallback; } if (runningCallsCount == 0 && idleCallback != null) { idleCallback.run(); } }复制代码
finished方法将此次请求从runningAsyncCalls移除后还执行了promoteCalls方法:
private void promoteCalls() { if (runningAsyncCalls.size() >= maxRequests) return; // Already running max capacity. if (readyAsyncCalls.isEmpty()) return; // No ready calls to promote. for (Iteratori = readyAsyncCalls.iterator(); i.hasNext(); ) { AsyncCall call = i.next(); if (runningCallsForHost(call) < maxRequestsPerHost) { i.remove(); runningAsyncCalls.add(call); executorService().execute(call); } if (runningAsyncCalls.size() >= maxRequests) return; // Reached max capacity. } }复制代码
最关键的一点就是会从readyAsyncCalls取出下一个请求,加入runningAsyncCalls中并交由线程池处理。
回到AsyncCall的execute方法:@Override protected void execute() { boolean signalledCallback = false; try { 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 { responseCallback.onFailure(RealCall.this, e); } } finally { client.dispatcher().finished(this); } }复制代码
从上面的代码看出,getResponseWithInterceptorChain方法返回了Response,这里是真正请求网络的地方。
2.2 回答一些问题
- OkHttp如何实现同步异步请求的? 发送的同步/异步请求都会在dispatcher中管理器状态
- 到底什么是Dispatcher Dispatcher的作用是维护请求的状态,并维护一个线程池,用于执行请求
- 异步请求为什么需要两个队列 Dispatcher:生产者 ExecutorService:消费者池 Deque:缓存 Deque:正在运行的任务
- Call执行完肯定需要在runningAsyncCalls队列中移除这个线程,那么readyAsyncCalls队列中的线程在什么时候才会被执行呢 从上面的源码分析中可以看到,无论网络请求是否成功,都会在finally 代码块中执行dispatcher的finished方法,然后再finished方法中判断,是否满足最大并发数小于等于64,最大主机请求数小于等于5的条件,如果满足,则从等待队列移除,加入到运行队列中,并执行这个请求。
3 Interceptor拦截器
拦截器是一种能够监控、重写、重试调用的机制。通常情况下,拦截器用来添加、移除、转换请求和响应的头部信息。比如将域名替换为IP地址,在请求头中添加host属性。
3.1 getResponseWithInterceptorChain
getResponseWithInterceptorChain的代码:
private Response getResponseWithInterceptorChain() throws IOException { // Build a full stack of interceptors. Listinterceptors = new ArrayList<>(); interceptors.addAll(client.interceptors()); interceptors.add(retryAndFollowUpInterceptor); interceptors.add(new BridgeInterceptor(client.cookieJar())); interceptors.add(new CacheInterceptor(client.internalCache())); interceptors.add(new ConnectInterceptor(client)); if (!retryAndFollowUpInterceptor.isForWebSocket()) { interceptors.addAll(client.networkInterceptors()); } interceptors.add(new CallServerInterceptor( retryAndFollowUpInterceptor.isForWebSocket())); Interceptor.Chain chain = new RealInterceptorChain( interceptors, null, null, null, 0, originalRequest); return chain.proceed(originalRequest); }复制代码
从这里可以看出,构建里一个拦截器的列表,并且执行了RealInterceptorChain.proceed方法,chain.proceed方法代码:
public Response proceed(Request request, StreamAllocation streamAllocation, HttpStream httpStream, Connection connection) throws IOException { ... // Call the next interceptor in the chain. RealInterceptorChain next = new RealInterceptorChain( interceptors, streamAllocation, httpStream, connection, index + 1, request); Interceptor interceptor = interceptors.get(index); Response response = interceptor.intercept(next); ... return response; }复制代码
proceed方法每次从拦截器列表中取出拦截器。当存在多个拦截器时,会等待下一个拦截器的调用返回。
proceed方法会依次执行拦截器的intercept方法,以下是各个拦截器的源码分析。3.2 RetryAndFollowUpInterceptor
RetryAndFollowUpInterceptor的intercept方法:
@Override public Response intercept(Chain chain) throws IOException { Request request = chain.request(); streamAllocation = new StreamAllocation( client.connectionPool(), createAddress(request.url())); int followUpCount = 0; Response priorResponse = null; while (true) { ... response = ((RealInterceptorChain) chain).proceed(request, streamAllocation, null, null); releaseConnection = false; ... // Attach the prior response if it exists. Such responses never have a body. if (priorResponse != null) { response = response.newBuilder() .priorResponse(priorResponse.newBuilder() .body(null) .build()) .build(); } Request followUp = followUpRequest(response); ... if (++followUpCount > MAX_FOLLOW_UPS) { streamAllocation.release(); throw new ProtocolException("Too many follow-up requests: " + followUpCount); } ... request = followUp; priorResponse = response; } }复制代码
RetryAndFollowUpInterceptor主要做了以下几件事:
- 创建StreamAllocation对象
- 调用((RealInterceptorChain) chain).proceed方法,执行下一个拦截器的操作
- 根据异常结果或者响应结果判断是否进行重新请求,重新请求最大次数为20次。
- 获取到下一次拦截器的response后,对response进行处理,并返回给上一个拦截器
3.3 BridgeInterceptor
BridgeInterceptor的intercept方法:
@Override 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) { requestBuilder.header("Content-Length", Long.toString(contentLength)); requestBuilder.removeHeader("Transfer-Encoding"); } else { requestBuilder.header("Transfer-Encoding", "chunked"); requestBuilder.removeHeader("Content-Length"); } } ... 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); responseBuilder.body(new RealResponseBody(strippedHeaders, Okio.buffer(responseBody))); } return responseBuilder.build(); }复制代码
BridgeInterceptor主要做了以下几件事:
- 负责将用户构建的一个Request请求转化为能够进行网络访问的请求,如添加请求头等相关信息
- 将这个符合网络请求的Request提交给下一个拦截器进行网络请求
- 将网络请求回来的响应Response转化为用户可用的Response,如Gzip解压等
3.3 CacheInterceptor
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(); Request networkRequest = strategy.networkRequest; Response cacheResponse = strategy.cacheResponse; if (cache != null) { cache.trackResponse(strategy); } 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) { return new Response.Builder() .request(chain.request()) .protocol(Protocol.HTTP_1_1) .code(504) .message("Unsatisfiable Request (only-if-cached)") .body(EMPTY_BODY) .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 (validate(cacheResponse, networkResponse)) { Response response = cacheResponse.newBuilder() .headers(combine(cacheResponse.headers(), networkResponse.headers())) .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()); } }复制代码
CacheInterceptor主要做了以下几件事:
- 通过Cache获取缓存,Cache类是进行缓存的读取和存取操作的,Cache以url的MD5为key,只缓存GET请求,通过DiskLruCache类,进行缓存。
- 根据CacheStrategy处理得到的networkRequest和cacheResponse进行判断处理,如果两个都为null,也就是不进行网络请求并且缓存不存在networkRequest或者过期,则返回504错误;当只有networkRequest为null时也就是不进行网络请求,如果有有效缓存则直接返回缓存;其他情况则请求网络
- 在判断为需要请求网络的时候,将请求传递给下一个拦截器进行网络请求
- 解析HTTP响应报头,如果有缓存并且可用,则用缓存的数据并更新缓存,否则就用网络请求返回的数据。 如何判断缓存是否有效?如果缓存有效的,服务器则返回304 Not Modified,否则直接返回body。如果缓存过期或者强制放弃缓存,则缓存策略全部交给服务器判断,客户端只需要发送条件GET请求即可。条件GET请求有两种方式:一种是Last-Modified-Date,另一种是ETag。这里采用了Last-Modified-Date,通过缓存和网络请求响应中的Last-Modified来计算是否是最新数据。如果是,则缓存有效。
3.4 ConnectInterceptor
ConnectInterceptor的intercept方法:
@Override public Response intercept(Chain chain) throws IOException { RealInterceptorChain realChain = (RealInterceptorChain) chain; Request request = realChain.request(); StreamAllocation streamAllocation = realChain.streamAllocation(); // We need the network to satisfy this request. Possibly for validating a conditional GET. boolean doExtensiveHealthChecks = !request.method().equals("GET"); HttpStream httpStream = streamAllocation.newStream(client, doExtensiveHealthChecks); RealConnection connection = streamAllocation.connection(); return realChain.proceed(request, streamAllocation, httpStream, connection); }复制代码
ConnectInterceptor主要做了两件事:
- ConnectInterceptor获取Interceptor传过来的StreamAllocation,通过streamAllocation.newStream()方法获取到HttpStream,HttpStream是用来处理request和response的
- 将刚才创建的用于网络IO的RealConnection对象,以及对于与服务器交互最为关键的HttpStream等对象传递给后面的拦截器 接下来看streamAllocation.newStream方法:
public HttpStream newStream(OkHttpClient client, boolean doExtensiveHealthChecks) { ... RealConnection resultConnection = findHealthyConnection(connectTimeout, readTimeout, writeTimeout, connectionRetryEnabled, doExtensiveHealthChecks); HttpStream resultStream; ... resultStream = new Http2xStream(client, this, resultConnection.framedConnection); ... resultStream = new Http1xStream( client, this, resultConnection.source, resultConnection.sink); ... return resultStream; ... }复制代码
newStream方法就是通过findHealthyConnection方法获取到健康的连接RealConnection,和根据Http1.0还是Http2.0协议创建不同的HttpStream。
下面是findHealthyConnection方法:/** * Finds a connection and returns it if it is healthy. If it is unhealthy the process is repeated * until a healthy connection is found. */ private RealConnection findHealthyConnection(int connectTimeout, int readTimeout, int writeTimeout, boolean connectionRetryEnabled, boolean doExtensiveHealthChecks) throws IOException { while (true) { RealConnection candidate = findConnection(connectTimeout, readTimeout, writeTimeout, connectionRetryEnabled); // 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; } }复制代码
findHealthyConnection通过findConnection方法循环遍历连接池,如果是新的连接,则直接使用此连接,如果是健康的连接,比如socket被关闭,或者输入输出流被关闭,则释放资源,继续遍历连接池。
下面是findConnection方法:/** * 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, boolean connectionRetryEnabled) throws IOException { ... // Attempt to get a connection from the pool. RealConnection pooledConnection = Internal.instance.get(connectionPool, address, this); if (pooledConnection != null) { this.connection = pooledConnection; return pooledConnection; } ... RealConnection newConnection = new RealConnection(selectedRoute); acquire(newConnection); synchronized (connectionPool) { Internal.instance.put(connectionPool, newConnection); this.connection = newConnection; if (canceled) throw new IOException("Canceled"); } newConnection.connect(connectTimeout, readTimeout, writeTimeout, address.connectionSpecs(), connectionRetryEnabled); routeDatabase().connected(newConnection.route()); return newConnection; }复制代码
findConnection首先会去尝试从连接此中,根据主机地址,尝试复用连接,如果不存在能复用的连接,则创建新的连接,放入连接池中,然后新的连接进行connect操作,创建Tunneled连接或者Socket连接,通过Okio进行流的操作。也就是说,OkHttp为了解决TCP3次握手4次挥手的效率问题,通过HTTP的keepalive connections的机制,使用了连接池复用连接。
3.5 OkHttp复用连接池
前面connectionPool复用连接池,现在看下connectionPool的具体代码
/** * Background threads are used to cleanup expired connections. There will be at most a single * thread running per connection pool. The thread pool executor permits the pool itself to be * garbage collected. */ private static final Executor executor = new ThreadPoolExecutor(0 /* corePoolSize */, Integer.MAX_VALUE /* maximumPoolSize */, 60L /* keepAliveTime */, TimeUnit.SECONDS, new SynchronousQueue(), Util.threadFactory("OkHttp ConnectionPool", true)); /** The maximum number of idle connections for each address. */ private final int maxIdleConnections; private final long keepAliveDurationNs; private final Deque connections = new ArrayDeque<>(); final RouteDatabase routeDatabase = new RouteDatabase(); boolean cleanupRunning;复制代码
连接池的主要变量:
- executor线程池,用于连接池的清理回收
- Deque,双向队列,双端队列同时具有队列和栈性质,经常在缓存中被使用,里面维护了RealConnection也就是socket物理连接的包装
- RouteDatabase,用来记录连接失败的路线名单,当连接失败的时候就会把失败的线路加进去。 以下是ConnectionPool的构造方法:
/** * Create a new connection pool with tuning parameters appropriate for a single-user application. * The tuning parameters in this pool are subject to change in future OkHttp releases. Currently * this pool holds up to 5 idle connections which will be evicted after 5 minutes of inactivity. */ public ConnectionPool() { this(5, 5, TimeUnit.MINUTES); } public ConnectionPool(int maxIdleConnections, long keepAliveDuration, TimeUnit timeUnit) { this.maxIdleConnections = maxIdleConnections; this.keepAliveDurationNs = timeUnit.toNanos(keepAliveDuration); // Put a floor on the keep alive duration, otherwise cleanup will spin loop. if (keepAliveDuration <= 0) { throw new IllegalArgumentException("keepAliveDuration <= 0: " + keepAliveDuration); } }复制代码
通过构造方法可以看出ConnectionPool默认空闲的socket最大连接数为5个,socket的keepAlive时间为5分钟。ConnectionPool是在OkHttpClient实例化时创建。
3.5.1 ConnectionPool的put方法:
void put(RealConnection connection) { assert (Thread.holdsLock(this)); if (!cleanupRunning) { cleanupRunning = true; executor.execute(cleanupRunnable); } connections.add(connection); }复制代码
先从连接池中清理回收空闲连接,再添加连接。
3.5.2 ConnectionPool的get方法:
/** Returns a recycled connection to {@code address}, or null if no such connection exists. */ RealConnection get(Address address, StreamAllocation streamAllocation) { assert (Thread.holdsLock(this)); for (RealConnection connection : connections) { if (connection.allocations.size() < connection.allocationLimit && address.equals(connection.route().address) && !connection.noNewStreams) { streamAllocation.acquire(connection); return connection; } } return null; }复制代码
遍历connections缓存列表。当某个连接计数的次数小于限制的大小,并且request的地址和缓存列表中此连接的地址完全匹配时,则直接复用缓存列表中的connection作为request的连接。
3.5.3 ConnectionPool的cleanupRunnable:
private final Runnable cleanupRunnable = new Runnable() { @Override public void run() { while (true) { 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) { } } } } } };复制代码
线程不断地调用cleanup方法来进行清理,并返回下次需要清理的间隔时间,然后调用wait方法进行等待一释放锁与时间片。当等待时间到了后,再次进行清理,并返回下次要清理的间隔时间。如此循环下去。
3.5.4 ConnectionPool的cleanup方法:
/** * Performs maintenance on this pool, evicting the connection that has been idle the longest if * either it has exceeded the keep alive limit or the idle connections limit. * *Returns the duration in nanos to sleep until the next scheduled call to this method. Returns * -1 if no further cleanups are required. */ 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
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; } }复制代码
cleanup方法主要做了以下几件事:
- 根据连接中的引用计数来计算空闲连接和活跃连接数,然后标记出空闲的连接
- 如果空闲连接keepAlive时间超过5分钟,或者空闲连接数超过5个,则从Deque中移除此连接
- 根据空闲连接或者活跃连接来返回下次需要清理的时间数:如果空闲连接大于0,则返回此连接即将到期的时间;如果都是活跃连接并且大于0,则返回默认的keepAlive时间5分钟;如果没有任何连接,则跳出循环并返回-1.
3.5.5 ConnectionPool的pruneAndGetAllocationCount方法:
/** * Prunes any leaked allocations and then returns the number of remaining live allocations on * {@code connection}. Allocations are leaked if the connection is tracking them but the * application code has abandoned them. Leak detection is imprecise and relies on garbage * collection. */ private int pruneAndGetAllocationCount(RealConnection connection, long now) { List> references = connection.allocations; for (int i = 0; i < references.size(); ) { Reference reference = references.get(i); if (reference.get() != null) { i++; continue; } // We've discovered a leaked allocation. This is an application bug. Platform.get().log(WARN, "A connection to " + connection.route().address().url() + " was leaked. Did you forget to close a response body?", null); 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(); }复制代码
pruneAndGetAllocationCount方法首先遍历传进来的RealConnection的StreamAllocation列表。如果StreamAllocation = null,则从列表中移除。如果列比为空,则说明连接没有引用了,则返回0,表示此连接是空闲连接;否则返回列表的数量,表示连接的活跃数量。
3.5.6 引用计数
/** * Use this allocation to hold {@code connection}. Each call to this must be paired with a call to * {@link #release} on the same connection. */ public void acquire(RealConnection connection) { connection.allocations.add(new WeakReference<>(this)); } /** Remove this allocation from the connection's list of allocations. */ private void release(RealConnection connection) { for (int i = 0, size = connection.allocations.size(); i < size; i++) { Referencereference = connection.allocations.get(i); if (reference.get() == this) { connection.allocations.remove(i); return; } } throw new IllegalStateException(); }复制代码
OkHttp回收连接使用了类似GC的引用计数算法,跟踪socket流的调用。这里的计数对象是StreamAllocation,它被反复执行acquire与release操作,这两个方法其实是在改变RealConnection中的List<Reference>的大小。
RealConnection是socket物理连接的包装,里面维护了List<Reference>的引用。List中StreamAllocation的数量也就是socket被引用的计数。如果计数为0,则表示连接没有被使用,空闲状态,需要回收;如果不为0,则表示仍在引用,不能关闭连接。3.5.7 小结
连接池复用的核心就是用Deque来存储连接,通过put、get、connectionBecameIdle和evictAll几个操作来对Deque进行操作,另外通过判断连接中的计数对象StreamAllocation来进行自动回收连接。
3.6 CallServerInterceptor
CallServerInterceptor的intercept方法:
@Override public Response intercept(Chain chain) throws IOException { HttpStream httpStream = ((RealInterceptorChain) chain).httpStream(); StreamAllocation streamAllocation = ((RealInterceptorChain) chain).streamAllocation(); Request request = chain.request(); long sentRequestMillis = System.currentTimeMillis(); httpStream.writeRequestHeaders(request); if (HttpMethod.permitsRequestBody(request.method()) && request.body() != null) { Sink requestBodyOut = httpStream.createRequestBody(request, request.body().contentLength()); BufferedSink bufferedRequestBody = Okio.buffer(requestBodyOut); request.body().writeTo(bufferedRequestBody); bufferedRequestBody.close(); } httpStream.finishRequest(); Response response = httpStream.readResponseHeaders() .request(request) .handshake(streamAllocation.connection().handshake()) .sentRequestAtMillis(sentRequestMillis) .receivedResponseAtMillis(System.currentTimeMillis()) .build(); if (!forWebSocket || response.code() != 101) { response = response.newBuilder() .body(httpStream.openResponseBody(response)) .build(); } if ("close".equalsIgnoreCase(response.request().header("Connection")) || "close".equalsIgnoreCase(response.header("Connection"))) { streamAllocation.noNewStreams(); } int code = response.code(); if ((code == 204 || code == 205) && response.body().contentLength() > 0) { throw new ProtocolException( "HTTP " + code + " had non-zero Content-Length: " + response.body().contentLength()); } return response; }复制代码
主要是根据前面拦截器处理的HttpStream、StreamAllocation以及Request进行实际的网络请求,以及获取response。
4. OkHttp网络请求过程的总结
- Call对象对请求进行封装
- dispatcher对请求进行分发
- getResponseWithInterceptors()方法,创建拦截器链
- 执行RetryAndFollowUpInterceptor,对请求进行重试重定向
- 执行BridgeInterceptor,对我们使用的OkHttp的request和response与HTTP协议中的request和response进行转换
- 执行CacheInterceptor,对请求进行判断,是否需要使用缓存
- 执行ConnectInterceptor,负责建立连接和流对象
- 执行CallServerInterceptor,负责最终实际的网络请求,主要进行发送请求和读取响应的操作。
参考文献:
Android进阶之光