手撕 Volley（二）
2016-10-05 本文已影响2842人
SavySoda
rick-and-morty-9.png
来继续我们的源码之旅，这边文章将会包括一下内容
添加请求——RequestQueue 的 add 方法
缓存调度——CacheDispacher 的 run 方法
网络调度——NetWorkDispacher 的 run 方法
网络请求——BasicNetwork 的 performRequest 方法
添加请求
前面说到了 Volley 的入口是创建一个 RequestQueue 队列，然后开启一个缓存线程和一组网络线程，等待用户 add 新的 request。那我们现在看一下 add 方法里面，RequestQueue 做了哪些事情。
/**
* Adds a Request to the dispatch queue.
* @param request The request to service
* @return The passed-in request
*/
public <T> Request<T> add(Request<T> request) {
// Tag the request as belonging to this queue and add it to the set of current requests.
request.setRequestQueue(this);
synchronized (mCurrentRequests) {
mCurrentRequests.add(request);
}
// Process requests in the order they are added.
request.setSequence(getSequenceNumber());
request.addMarker("add-to-queue");
// If the request is uncacheable, skip the cache queue and go straight to the network.
if (!request.shouldCache()) {
mNetworkQueue.add(request);
return request;
}
// Insert request into stage if there's already a request with the same cache key in flight.
synchronized (mWaitingRequests) {
String cacheKey = request.getCacheKey();
if (mWaitingRequests.containsKey(cacheKey)) {
// There is already a request in flight. Queue up.
Queue<Request<?>> stagedRequests = mWaitingRequests.get(cacheKey);
if (stagedRequests == null) {
stagedRequests = new LinkedList<Request<?>>();
}
stagedRequests.add(request);
mWaitingRequests.put(cacheKey, stagedRequests);
if (VolleyLog.DEBUG) {
VolleyLog.v("Request for cacheKey=%s is in flight, putting on hold.", cacheKey);
}
} else {
// Insert 'null' queue for this cacheKey, indicating there is now a request in
// flight.
mWaitingRequests.put(cacheKey, null);
mCacheQueue.add(request);
}
return request;
}
}
这是一个泛型方法，来看一下他的流程图
add_flow.png
被添加到缓存队列中的 Request 就可以去缓存里面进行缓存调度查找匹配了。先看刚才流程，RquestQueue、Cache、mCurrentRequests、mWaitingRequests 手撕Volley（一）类图有介绍，那么疑问来了：
- Request 是啥
- cacheKey 是怎么生成的
/**
* Base class for all network requests.
*
* @param <T> The type of parsed response this request expects.
*/
public abstract class Request<T> implements Comparable<Request<T>> {
public Request(int method, String url, Response.ErrorListener listener) {
mMethod = method;
mUrl = url;
mIdentifier = createIdentifier(method, url);
mErrorListener = listener;
setRetryPolicy(new DefaultRetryPolicy());
mDefaultTrafficStatsTag = findDefaultTrafficStatsTag(url);
}
首先可以看到 Request 是一个抽象类，是所有请求的基类。
来看 Requst 的构造器，method、url 分别对应 Http 协议报文里面的 method 和 url。HTTP 协议就不展开说了，细节还是要看我手撕Volley（一）里面HTTP协议相关的的内容。
private static String createIdentifier(final int method, final String url) {
return InternalUtils.sha1Hash("Request:" + method + ":" + url +
":" + System.currentTimeMillis() + ":" + (sCounter++));
}
而 mIdentifier 是根据当前毫秒和 method 以及 url 计算的哈希作为唯一标识。
/** Default tag for {@link TrafficStats}. */
private final int mDefaultTrafficStatsTag;
/**
* @return The hashcode of the URL's host component, or 0 if there is none.
*/
private static int findDefaultTrafficStatsTag(String url) {
if (!TextUtils.isEmpty(url)) {
Uri uri = Uri.parse(url);
if (uri != null) {
String host = uri.getHost();
if (host != null) {
return host.hashCode();
}
}
}
return 0;
}
mDefaultTrafficStatsTag 是 host （域名）的一个哈希，有啥用暂时未知。
/** The retry policy for this request. */
private RetryPolicy mRetryPolicy;
/**
* Retry policy for a request.
*/
public interface RetryPolicy {
/**
* Returns the current timeout (used for logging).
*/
public int getCurrentTimeout();
/**
* Returns the current retry count (used for logging).
*/
public int getCurrentRetryCount();
/**
* Prepares for the next retry by applying a backoff to the timeout.
* @param error The error code of the last attempt.
* @throws VolleyError In the event that the retry could not be performed (for example if we
* ran out of attempts), the passed in error is thrown.
*/
public void retry(VolleyError error) throws VolleyError;
}
RetryPolicy 也是一个接口，定义了默认超时时间以及重连次数。他的默认实现是 DefaultRetryPolicy，里面定义了几个常量当作默认实现。
/** The default socket timeout in milliseconds */
public static final int DEFAULT_TIMEOUT_MS = 2500;
/** The default number of retries */
public static final int DEFAULT_MAX_RETRIES = 0;
/** The default backoff multiplier */
public static final float DEFAULT_BACKOFF_MULT = 1f;
最后，附上类图：
request.png
retry_policy.png
到了这里，add 方法我们就基本理解了，刚才说到，add 方法的最后 request 被添加到 缓存队列里面去匹配，那下面就来看缓存队列里做了什么
缓存调度
还记的前面说过 CacheDispacher 继承了 Thread，是一个线程类，他的 run 方法是一个 while true 死循环，有一个标记位 mQuit 来退出循环。
先看成员变量
private static final boolean DEBUG = VolleyLog.DEBUG;
/** 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;
CacheDispather.png
- mCacheQueue 和 mNetworkQueue
是阻塞队列 - ResponseDelivery 接口用来 post response 或者 error
终于 Volley 的核心代码之一
@Override
public void run() {
if (DEBUG) VolleyLog.v("start new dispatcher");
Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND);
// Make a blocking call to initialize the cache.
mCache.initialize();
Request<?> request;
while (true) {
// release previous request object to avoid leaking request object when mQueue is drained.
request = null;
try {
// Take a request from the queue.
request = mCacheQueue.take();
} catch (InterruptedException e) {
// We may have been interrupted because it was time to quit.
if (mQuit) {
return;
}
continue;
}
try {
request.addMarker("cache-queue-take");
// If the request has been canceled, don't bother dispatching it.
if (request.isCanceled()) {
request.finish("cache-discard-canceled");
continue;
}
// 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.
mNetworkQueue.put(request);
continue;
}
// If it is completely expired, just send it to the network.
if (entry.isExpired()) {
request.addMarker("cache-hit-expired");
request.setCacheEntry(entry);
mNetworkQueue.put(request);
continue;
}
// 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 (!entry.refreshNeeded()) {
// 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;
// Post the intermediate response back to the user and have
// the delivery then forward the request along to the network.
final Request<?> finalRequest = request;
mDelivery.postResponse(request, response, new Runnable() {
@Override
public void run() {
try {
mNetworkQueue.put(finalRequest);
} catch (InterruptedException e) {
// Not much we can do about this.
}
}
});
}
} catch (Exception e) {
VolleyLog.e(e, "Unhandled exception %s", e.toString());
}
}
}
run方法流程图如下：
cachedispacher_run.png
我们注意到
- continue 大量使用
- log 信息记录很详细
/**
* Data and metadata for an entry returned by the cache.
*/
public static 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;
/** Soft TTL for this record. */
public long softTtl;
/** Immutable response headers as received from server; must be non-null. */
public Map<String, String> responseHeaders = Collections.emptyMap();
/** True if the entry is expired. */
public boolean isExpired() {
return this.ttl < System.currentTimeMillis();
}
/** True if a refresh is needed from the original data source. */
public boolean refreshNeeded() {
return this.softTtl < System.currentTimeMillis();
}
}
这里再贴一次 Cahce 接口的内部类 Entry，因为他真的太重要了，我们看连个判断过期和需要刷新的方法分别是，两个成员变量跟当前时间的对比。而 data 是二进制数组，我们都知道在 HTTP 中 start line 和 headers 是明文存储的，而 Entity 是没有规定的，一般我们都用二进制流传输，可以减少传输流量，并且安全，data 这里就是用来保存 Entity 的。
Cache 的默认实现是 DiskBasedCache，
* Cache implementation that caches files directly onto the hard disk in the specified
* directory. The default disk usage size is 5MB, but is configurable.
*/
public class DiskBasedCache implements Cache {
/** Map of the Key, CacheHeader pairs */
private final Map<String, CacheHeader> mEntries =
new LinkedHashMap<String, CacheHeader>(16, .75f, true);
可以看到默认大小为5M，但是可以自己配置，内部用了一个 LinkedHashMap 来保存 request 的 CacheHeader ，CacheHeader 是为了存储 Entity 中的 header 和 data 的 size，我觉得是为了避免存大量的 data 吧。
LinkedHashMap 是为了实现 FIFO 的缓存替换策略，我们知道，在空间不足时向 HashMap 中 put 数据就需要删除一些内容用来保证最新 put数据的成功。
/**
* Prunes the cache to fit the amount of bytes specified.
* @param neededSpace The amount of bytes we are trying to fit into the cache.
*/
private void pruneIfNeeded(int neededSpace) {
if ((mTotalSize + neededSpace) < mMaxCacheSizeInBytes) {
return;
}
if (VolleyLog.DEBUG) {
VolleyLog.v("Pruning old cache entries.");
}
long before = mTotalSize;
int prunedFiles = 0;
long startTime = SystemClock.elapsedRealtime();
Iterator<Map.Entry<String, CacheHeader>> iterator = mEntries.entrySet().iterator();
while (iterator.hasNext()) {
Map.Entry<String, CacheHeader> entry = iterator.next();
CacheHeader e = entry.getValue();
boolean deleted = getFileForKey(e.key).delete();
if (deleted) {
mTotalSize -= e.size;
} else {
VolleyLog.d("Could not delete cache entry for key=%s, filename=%s",
e.key, getFilenameForKey(e.key));
}
iterator.remove();
prunedFiles++;
if ((mTotalSize + neededSpace) < mMaxCacheSizeInBytes * HYSTERESIS_FACTOR) {
break;
}
}
if (VolleyLog.DEBUG) {
VolleyLog.v("pruned %d files, %d bytes, %d ms",
prunedFiles, (mTotalSize - before), SystemClock.elapsedRealtime() - startTime);
}
}
pruneIfNeeded 方法是在 put 方法的第一行执行的，做的就是这件事，Volley 并没有使用 LRU。而是使用的 FIFO。DiskBasedCache 剩下的就是一些文件操作了，就不挨着看了。
网络调度
同缓存调度，NetworkDispatcher 也是一个 Thread 子类，主要看它的成员变量和 run 方法，说干就干
public class NetworkDispatcher extends Thread {
/** The queue of requests to service. */
private final BlockingQueue<Request<?>> mQueue;
/** The network interface for processing requests. */
private final Network mNetwork;
/** The cache to write to. */
private final Cache mCache;
/** For posting responses and errors. */
private final ResponseDelivery mDelivery;
/** Used for telling us to die. */
private volatile boolean mQuit = false;
NetworkDispatcher 类图
NetworkDispatcher .png
- 一个阻塞队列
- 一个 NetWork 接口
- 一个 Cache 接口
- 一个结果分发器
好了师徒四人凑齐了，可以去取经了。开个玩笑，以上四种类型前面手撕Volley（一）介绍，忘记的可以去前面查，这里就不再介绍了。再看 run 方法。
@Override
public void run() {
Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND);
Request<?> request;
while (true) {
long startTimeMs = SystemClock.elapsedRealtime();
// release previous request object to avoid leaking request object when mQueue is drained.
request = null;
try {
// Take a request from the queue.
request = mQueue.take();
} catch (InterruptedException e) {
// We may have been interrupted because it was time to quit.
if (mQuit) {
return;
}
continue;
}
try {
request.addMarker("network-queue-take");
// If the request was cancelled already, do not perform the
// network request.
if (request.isCanceled()) {
request.finish("network-discard-cancelled");
continue;
}
addTrafficStatsTag(request);
// Perform the network request.
NetworkResponse networkResponse = mNetwork.performRequest(request);
request.addMarker("network-http-complete");
// If the server returned 304 AND we delivered a response already,
// we're done -- don't deliver a second identical response.
if (networkResponse.notModified && request.hasHadResponseDelivered()) {
request.finish("not-modified");
continue;
}
// Parse the response here on the worker thread.
Response<?> response = request.parseNetworkResponse(networkResponse);
request.addMarker("network-parse-complete");
// 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");
}
// Post the response back.
request.markDelivered();
mDelivery.postResponse(request, response);
} catch (VolleyError volleyError) {
volleyError.setNetworkTimeMs(SystemClock.elapsedRealtime() - startTimeMs);
parseAndDeliverNetworkError(request, volleyError);
} catch (Exception e) {
VolleyLog.e(e, "Unhandled exception %s", e.toString());
VolleyError volleyError = new VolleyError(e);
volleyError.setNetworkTimeMs(SystemClock.elapsedRealtime() - startTimeMs);
mDelivery.postError(request, volleyError);
}
}
}
长的一匹，跟缓存调度很像，流程图如下：
networkdispatcher.png
Network的具体的实现之前已经分析是BasicNetwork，先看成员变量
/**
* A network performing Volley requests over an {@link HttpStack}.
*/
public class BasicNetwork implements Network {
protected static final boolean DEBUG = VolleyLog.DEBUG;
private static int SLOW_REQUEST_THRESHOLD_MS = 3000;
private static int DEFAULT_POOL_SIZE = 4096;
protected final HttpStack mHttpStack;
protected final ByteArrayPool mPool;
- 两个常量，分别表示最长请求时间和线程池大小
- 一个HttpStack 接口，真正执行网络请求的类
- 二进制数组池,一个工具类
类图如下
network.png
再看实现的 performRequest 方法
@Override
public NetworkResponse performRequest(Request<?> request) throws VolleyError {
long requestStart = SystemClock.elapsedRealtime();
while (true) {
HttpResponse httpResponse = null;
byte[] responseContents = null;
Map<String, String> responseHeaders = Collections.emptyMap();
try {
// Gather headers.
Map<String, String> headers = new HashMap<String, String>();
addCacheHeaders(headers, request.getCacheEntry());
httpResponse = mHttpStack.performRequest(request, headers);
StatusLine statusLine = httpResponse.getStatusLine();
int statusCode = statusLine.getStatusCode();
responseHeaders = convertHeaders(httpResponse.getAllHeaders());
// Handle cache validation.
if (statusCode == HttpStatus.SC_NOT_MODIFIED) {
Entry entry = request.getCacheEntry();
if (entry == null) {
return new NetworkResponse(HttpStatus.SC_NOT_MODIFIED, null,
responseHeaders, true,
SystemClock.elapsedRealtime() - requestStart);
}
// A HTTP 304 response does not have all header fields. We
// have to use the header fields from the cache entry plus
// the new ones from the response.
// http://www.w3.org/Protocols/rfc2616/rfc2616-sec10.html#sec10.3.5
entry.responseHeaders.putAll(responseHeaders);
return new NetworkResponse(HttpStatus.SC_NOT_MODIFIED, entry.data,
entry.responseHeaders, true,
SystemClock.elapsedRealtime() - requestStart);
}
// Handle moved resources
if (statusCode == HttpStatus.SC_MOVED_PERMANENTLY || statusCode == HttpStatus.SC_MOVED_TEMPORARILY) {
String newUrl = responseHeaders.get("Location");
request.setRedirectUrl(newUrl);
}
// Some responses such as 204s do not have content. We must check.
if (httpResponse.getEntity() != null) {
responseContents = entityToBytes(httpResponse.getEntity());
} else {
// Add 0 byte response as a way of honestly representing a
// no-content request.
responseContents = new byte[0];
}
// if the request is slow, log it.
long requestLifetime = SystemClock.elapsedRealtime() - requestStart;
logSlowRequests(requestLifetime, request, responseContents, statusLine);
if (statusCode < 200 || statusCode > 299) {
throw new IOException();
}
return new NetworkResponse(statusCode, responseContents, responseHeaders, false,
SystemClock.elapsedRealtime() - requestStart);
} catch (SocketTimeoutException e) {
attemptRetryOnException("socket", request, new TimeoutError());
} catch (ConnectTimeoutException e) {
attemptRetryOnException("connection", request, new TimeoutError());
} catch (MalformedURLException e) {
throw new RuntimeException("Bad URL " + request.getUrl(), e);
} catch (IOException e) {
int statusCode = 0;
NetworkResponse networkResponse = null;
if (httpResponse != null) {
statusCode = httpResponse.getStatusLine().getStatusCode();
} else {
throw new NoConnectionError(e);
}
if (statusCode == HttpStatus.SC_MOVED_PERMANENTLY ||
statusCode == HttpStatus.SC_MOVED_TEMPORARILY) {
VolleyLog.e("Request at %s has been redirected to %s", request.getOriginUrl(), request.getUrl());
} else {
VolleyLog.e("Unexpected response code %d for %s", statusCode, request.getUrl());
}
if (responseContents != null) {
networkResponse = new NetworkResponse(statusCode, responseContents,
responseHeaders, false, SystemClock.elapsedRealtime() - requestStart);
if (statusCode == HttpStatus.SC_UNAUTHORIZED ||
statusCode == HttpStatus.SC_FORBIDDEN) {
attemptRetryOnException("auth",
request, new AuthFailureError(networkResponse));
} else if (statusCode == HttpStatus.SC_MOVED_PERMANENTLY ||
statusCode == HttpStatus.SC_MOVED_TEMPORARILY) {
attemptRetryOnException("redirect",
request, new RedirectError(networkResponse));
} else {
// TODO: Only throw ServerError for 5xx status codes.
throw new ServerError(networkResponse);
}
} else {
throw new NetworkError(e);
}
}
}
}
流程图如下：
network_performrequest
最终的网络执行还是在HTTPStack中，待续。。
