Volley之CacheDispatcher缓存
在源码解析Volley请求队列RequestQueue一文最后，我们看了向请求队列添加Request网络请求的逻辑。
<T> void beginRequest(Request<T> request) {
// If the request is uncacheable, skip the cache queue and go straight to the network.
if (!request.shouldCache()) {
sendRequestOverNetwork(request);
} else {
mCacheQueue.add(request);
}
}如果请求开启缓存，那么会先将请求放到缓存队列中处理，缓存队列没有或者缓存过期的话再放到网络请求队列中处理并返回。
1、CacheDispatcher
CacheDispatcher也是继承自Thread。
1.1、CacheDispatcher初始化
在RequestQueue的start()方法中创建一个缓存调度程序CacheDispatcher，以及默认4个网络请求调度程序NetworkDispatcher。
/** Starts the dispatchers in this queue. */
public void start() {
stop(); // Make sure any currently running dispatchers are stopped.
// Create the cache dispatcher and start it.
mCacheDispatcher = new CacheDispatcher(mCacheQueue, mNetworkQueue, mCache, mDelivery);
mCacheDispatcher.start();
// Create network dispatchers (and corresponding threads) up to the pool size.
for (int i = 0; i < mDispatchers.length; i++) {
NetworkDispatcher networkDispatcher =
new NetworkDispatcher(mNetworkQueue, mNetwork, mCache, mDelivery);
mDispatchers[i] = networkDispatcher;
networkDispatcher.start();
}
}CacheDispatcher的构造函数如下，将RequestQueue中的全局变量：缓存队列、网络请求队列、缓存、结果回调分发器，传入CacheDispatcher中处理缓存使用。
/**
* Creates a new cache triage dispatcher thread. You must call {@link #start()} in order to
* begin processing.
*
* @param cacheQueue Queue of incoming requests for triage
* @param networkQueue Queue to post requests that require network to
* @param cache Cache interface to use for resolution
* @param delivery Delivery interface to use for posting responses
*/
public CacheDispatcher(
BlockingQueue<Request<?>> cacheQueue,
BlockingQueue<Request<?>> networkQueue,
Cache cache,
ResponseDelivery delivery) {
mCacheQueue = cacheQueue;
mNetworkQueue = networkQueue;
mCache = cache;
mDelivery = delivery;
mWaitingRequestManager = new WaitingRequestManager(this, networkQueue, delivery);
}1.2、CacheDispatcher成员
CacheDispatcher类前4个成员都是从RequestQueue初始化获得：
/** The queue of requests coming in for triage. */
private final BlockingQueue<Request<?>> mCacheQueue;
/** The queue of requests going out to the network. */
private final BlockingQueue<Request<?>> mNetworkQueue;
/** The cache to read from. */
private final Cache mCache;
/** For posting responses. */
private final ResponseDelivery mDelivery;
/** Used for telling us to die. */
private volatile boolean mQuit = false;
/** Manage list of waiting requests and de-duplicate requests with same cache key. */
private final WaitingRequestManager mWaitingRequestManager;最后一个WaitingRequestManager类，用来处理缓存miss时，需要通过网络请求，网络结果返回时回调的处理。
1.3、启动CacheDispatcher
重写Thread的run()方法，开启无限循环，从阻塞队列中取数据处理。在这里对Cache进行初始化
@Override
public void run() {
if (DEBUG) VolleyLog.v("start new dispatcher");
Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND);
// 进行阻塞调用，初始化缓存。
mCache.initialize();
while (true) {
try {
processRequest();
} catch (InterruptedException e) {
// We may have been interrupted because it was time to quit.
if (mQuit) {
Thread.currentThread().interrupt();
return;
}
VolleyLog.e(
"Ignoring spurious interrupt of CacheDispatcher thread; "
+ "use quit() to terminate it");
}
}
}1.4、终止CacheDispatcher
CacheDispatcher继承自线程，quit()方法中就是调用自身线程的interrupt()方法，并将退出标志位置为true。
/**
* Forces this dispatcher to quit immediately. If any requests are still in the queue, they are
* not guaranteed to be processed.
*/
public void quit() {
mQuit = true;
interrupt();
}2、缓存处理
启动缓存调度程序后，开启无线循环，processRequest()方法从mCacheQueue缓存队列中取出缓存请求处理。
2.1、二级调用processRequest()
先调用processRequest()方法，内部先从mCacheQueue缓存队列拿出一个Request实例
// Extracted to its own method to ensure locals have a constrained liveness scope by the GC.
// This is needed to avoid keeping previous request references alive for an indeterminate amount
// of time. Update consumer-proguard-rules.pro when modifying this. See also
// https://github.com/google/volley/issues/114
private void processRequest() throws InterruptedException {
// Get a request from the cache triage queue, blocking until
// at least one is available.
final Request<?> request = mCacheQueue.take();
processRequest(request);
}将缓存实例传递给processRequest(final Request<?> request)处理，完整方法一共93行：
@VisibleForTesting
void processRequest(final Request<?> request) throws InterruptedException {
request.addMarker("cache-queue-take");
request.sendEvent(RequestQueue.RequestEvent.REQUEST_CACHE_LOOKUP_STARTED);
try {
// If the request has been canceled, don't bother dispatching it.
if (request.isCanceled()) {
request.finish("cache-discard-canceled");
return;
}
// Attempt to retrieve this item from cache.
Cache.Entry entry = mCache.get(request.getCacheKey());
if (entry == null) {
request.addMarker("cache-miss");
// Cache miss; send off to the network dispatcher.
if (!mWaitingRequestManager.maybeAddToWaitingRequests(request)) {
mNetworkQueue.put(request);
}
return;
}
// Use a single instant to evaluate cache expiration. Otherwise, a cache entry with
// identical soft and hard TTL times may appear to be valid when checking isExpired but
// invalid upon checking refreshNeeded(), triggering a soft TTL refresh which should be
// impossible.
long currentTimeMillis = System.currentTimeMillis();
// If it is completely expired, just send it to the network.
if (entry.isExpired(currentTimeMillis)) {
request.addMarker("cache-hit-expired");
request.setCacheEntry(entry);
if (!mWaitingRequestManager.maybeAddToWaitingRequests(request)) {
mNetworkQueue.put(request);
}
return;
}
// We have a cache hit; parse its data for delivery back to the request.
request.addMarker("cache-hit");
Response<?> response =
request.parseNetworkResponse(
new NetworkResponse(entry.data, entry.responseHeaders));
request.addMarker("cache-hit-parsed");
if (!response.isSuccess()) {
request.addMarker("cache-parsing-failed");
mCache.invalidate(request.getCacheKey(), true);
request.setCacheEntry(null);
if (!mWaitingRequestManager.maybeAddToWaitingRequests(request)) {
mNetworkQueue.put(request);
}
return;
}
if (!entry.refreshNeeded(currentTimeMillis)) {
// Completely unexpired cache hit. Just deliver the response.
mDelivery.postResponse(request, response);
} else {
// Soft-expired cache hit. We can deliver the cached response,
// but we need to also send the request to the network for
// refreshing.
request.addMarker("cache-hit-refresh-needed");
request.setCacheEntry(entry);
// Mark the response as intermediate.
response.intermediate = true;
if (!mWaitingRequestManager.maybeAddToWaitingRequests(request)) {
// Post the intermediate response back to the user and have
// the delivery then forward the request along to the network.
mDelivery.postResponse(
request,
response,
new Runnable() {
@Override
public void run() {
try {
mNetworkQueue.put(request);
} catch (InterruptedException e) {
// Restore the interrupted status
Thread.currentThread().interrupt();
}
}
});
} else {
// request has been added to list of waiting requests
// to receive the network response from the first request once it returns.
mDelivery.postResponse(request, response);
}
}
} finally {
request.sendEvent(RequestQueue.RequestEvent.REQUEST_CACHE_LOOKUP_FINISHED);
}
}和NetworkDispatcher一样，每次处理请求都会先判断请求是否已被取消，在多处关键节点都会判断请求是否取消。
// If the request has been canceled, don't bother dispatching it.
if (request.isCanceled()) {
request.finish("cache-discard-canceled");
return;
}下面逐段分解该方法，看看如何处理缓存逻辑的：
2.2、缓存miss
从mCache缓存中取当前Request请求对应的Key，如果缓存中没有那么将其放入到网络请求队列mNetworkQueue中。Volley网络请求逻辑见另一篇博客Volley之网络请求NetworkDispatcher。
// Attempt to retrieve this item from cache.
Cache.Entry entry = mCache.get(request.getCacheKey());
if (entry == null) {
request.addMarker("cache-miss");
// Cache miss; send off to the network dispatcher.
if (!mWaitingRequestManager.maybeAddToWaitingRequests(request)) {
mNetworkQueue.put(request);
}
return;
}2.3、缓存失效
接着，如果有缓存，会判断缓存Cache.Entry是否有效。如果失效就将请求放入到网络请求队列中处理。并且将其放入WaitingRequestManager中标记等待网络请求结果返回后回调。
long currentTimeMillis = System.currentTimeMillis();
// If it is completely expired, just send it to the network.
if (entry.isExpired(currentTimeMillis)) {
request.addMarker("cache-hit-expired");
request.setCacheEntry(entry);
if (!mWaitingRequestManager.maybeAddToWaitingRequests(request)) {
mNetworkQueue.put(request);
}
return;
}Entry是定义在Cache中的缓存实体类，内部存放数据、最大生命有效期TTL等信息。
/** Data and metadata for an entry returned by the cache. */
class Entry {
/** The data returned from cache. */
public byte[] data;
/** ETag for cache coherency. */
public String etag;
/** Date of this response as reported by the server. */
public long serverDate;
/** The last modified date for the requested object. */
public long lastModified;
/** TTL for this record. */
public long ttl;
...
/** True if the entry is expired. */
public boolean isExpired() {
return isExpired(System.currentTimeMillis());
}
boolean isExpired(long currentTimeMillis) {
return this.ttl < currentTimeMillis;
}
...
}2.4、WaitingRequestManager
获取缓存失败都会将请求放入到WaitingRequestManager中管理，该类实现Request.NetworkRequestCompleteListener接口
/** Callback to notify when the network request returns. */
/* package */ interface NetworkRequestCompleteListener {
/** Callback when a network response has been received. */
void onResponseReceived(Request<?> request, Response<?> response);
/** Callback when request returns from network without valid response. */
void onNoUsableResponseReceived(Request<?> request);
}获取缓存失败的情况都会调用WaitingRequestManager的maybeAddToWaitingRequests(Request<?> request)方法，将其放入到等待请求队列中，等待网络结果请求返回回调处理缓存相关的逻辑：
synchronized boolean maybeAddToWaitingRequests(Request<?> request) {
String cacheKey = request.getCacheKey();
// Insert request into stage if there's already a request with the same cache key
// in flight.
if (mWaitingRequests.containsKey(cacheKey)) {
// There is already a request in flight. Queue up.
List<Request<?>> stagedRequests = mWaitingRequests.get(cacheKey);
if (stagedRequests == null) {
stagedRequests = new ArrayList<>();
}
request.addMarker("waiting-for-response");
stagedRequests.add(request);
mWaitingRequests.put(cacheKey, stagedRequests);
if (VolleyLog.DEBUG) {
VolleyLog.d("Request for cacheKey=%s is in flight, putting on hold.", cacheKey);
}
return true;
} else {
// Insert 'null' queue for this cacheKey, indicating there is now a request in
// flight.
mWaitingRequests.put(cacheKey, null);
request.setNetworkRequestCompleteListener(this);
if (VolleyLog.DEBUG) {
VolleyLog.d("new request, sending to network %s", cacheKey);
}
return false;
}
}如果请求已在该等待请求队列中，那么刷新。否者将当前请求加入到等待请求队列中。
3、回调处理
有两条回调路径：缓存命中回调 和 缓存miss后的网络请求结果回调。
3.1、缓存命中回调
①对于获取缓存后过期的缓存命中。只需提供回调响应即可。
if (!entry.refreshNeeded(currentTimeMillis)) {
// Completely unexpired cache hit. Just deliver the response.
mDelivery.postResponse(request, response);
}②将缓存命中的响应发回给用户并进行传递，然后再将请求转发到网络，这一步应该是为了刷新缓存。
if (!mWaitingRequestManager.maybeAddToWaitingRequests(request)) {
// Post the intermediate response back to the user and have
// the delivery then forward the request along to the network.
mDelivery.postResponse(
request,
response,
new Runnable() {
@Override
public void run() {
try {
mNetworkQueue.put(request);
} catch (InterruptedException e) {
// Restore the interrupted status
Thread.currentThread().interrupt();
}
}
});
}③该请求已添加到等待请求的列表中，以便在第一个请求返回时从第一个请求接收网络响应。
// request has been added to list of waiting requests
// to receive the network response from the first request once it returns.
mDelivery.postResponse(request, response);这三种都是通过ExecutorDelivery回调Request分发相应，见另一篇博客Volley之网络请求NetworkDispatcher。网络请求完毕之后再回到缓存回调中处理，继续3.2节代码解读。
3.2、网络请求结果回调：WaitingRequestManager
在Volley之网络请求NetworkDispatcher一文第三节网络结果回调中，在NetworkDispatcher的processRequest(Request<?> request)方法中有一步操作是将网络结果通过WaitingRequestManager回调通知CacheDispatcher。并且在这一步将响应写入缓存(需要的话)。
// Write to cache if applicable.
// TODO: Only update cache metadata instead of entire record for 304s.
if (request.shouldCache() && response.cacheEntry != null) {
mCache.put(request.getCacheKey(), response.cacheEntry);
request.addMarker("network-cache-written");
}
...
request.notifyListenerResponseReceived(response);调用Request中的NetworkRequestCompleteListener，这个监听器就是在前面缓存失效时给Request所设置的WaitingRequestManager，WaitingRequestManager类实现NetworkRequestCompleteListener接口。
/**
* Notify NetworkRequestCompleteListener that a valid response has been received which can be
* used for other, waiting requests.
*
* @param response received from the network
*/
/* package */ void notifyListenerResponseReceived(Response<?> response) {
NetworkRequestCompleteListener listener;
synchronized (mLock) {
listener = mRequestCompleteListener;
}
if (listener != null) {
listener.onResponseReceived(this, response);
}
}举个例子，网络请求结果返回，回调WaitingRequestManager的onResponseReceived(Request<?> request, Response<?> response)方法，最终通过ExecutorDelivery分发结果。
/** Request received a valid response that can be used by other waiting requests. */
@Override
public void onResponseReceived(Request<?> request, Response<?> response) {
if (response.cacheEntry == null || response.cacheEntry.isExpired()) {
onNoUsableResponseReceived(request);
return;
}
String cacheKey = request.getCacheKey();
List<Request<?>> waitingRequests;
synchronized (this) {
waitingRequests = mWaitingRequests.remove(cacheKey);
}
if (waitingRequests != null) {
if (VolleyLog.DEBUG) {
VolleyLog.v(
"Releasing %d waiting requests for cacheKey=%s.",
waitingRequests.size(), cacheKey);
}
// Process all queued up requests.
for (Request<?> waiting : waitingRequests) {
mResponseDelivery.postResponse(waiting, response);
}
}
}而对于网络请求结果失败、无可用相应内容，则回调WaitingRequestManager的onNoUsableResponseReceived(Request<?> request)方法：
/** No valid response received from network, release waiting requests. */
@Override
public synchronized void onNoUsableResponseReceived(Request<?> request) {
String cacheKey = request.getCacheKey();
List<Request<?>> waitingRequests = mWaitingRequests.remove(cacheKey);
if (waitingRequests != null && !waitingRequests.isEmpty()) {
if (VolleyLog.DEBUG) {
VolleyLog.v(
"%d waiting requests for cacheKey=%s; resend to network",
waitingRequests.size(), cacheKey);
}
Request<?> nextInLine = waitingRequests.remove(0);
mWaitingRequests.put(cacheKey, waitingRequests);
nextInLine.setNetworkRequestCompleteListener(this);
// RequestQueue will be non-null if this instance was created in AsyncRequestQueue.
if (mRequestQueue != null) {
// Will send the network request from the RequestQueue.
mRequestQueue.sendRequestOverNetwork(nextInLine);
} else if (mCacheDispatcher != null && mNetworkQueue != null) {
// If we're not using the AsyncRequestQueue, then submit it to the network queue.
try {
mNetworkQueue.put(nextInLine);
} catch (InterruptedException iex) {
VolleyLog.e("Couldn't add request to queue. %s", iex.toString());
// Restore the interrupted status of the calling thread (i.e. NetworkDispatcher)
Thread.currentThread().interrupt();
// Quit the current CacheDispatcher thread.
mCacheDispatcher.quit();
}
}
}
}Volley的CacheDispatcher缓存调度程序这一条线的逻辑梳理完了，大家从Volley的源码中也要领悟一些设计模式。优秀的框架设计不需要复杂的逻辑。
