Volley库源码分析(上):http://blog.csdn.net/mba16c35/article/details/43944703
Volley库源码分析(下):http://blog.csdn.net/mba16c35/article/details/44589137
整体框架
主线程和cache线程都只有一个,而NetworkDispatcher线程可以有多个,这样能解决比并行问题。如下图:
核心:NetworkDispatcher
关键步骤
request = mQueue.take();
2.发起网络请求
NetworkResponse networkResponse = mNetwork.performRequest(request);
Response<?> response = request.parseNetworkResponse(networkResponse);
4.将网络响应加入缓存
mCache.put(request.getCacheKey(), response.cacheEntry);
5.将网络响应发回主线程
mDelivery.postResponse(request, response);
下面是NetworkDispatcher线程的主要代码:
@Override public void run() { Process.setThreadPriority(Process.THREAD_PRIORITY_BACKGROUND); Request request; while (true) { try { // Take a request from the queue. request = mQueue.take();//1.从请求队列中取出一个网络请求,mQueue是BlockingQueue<Request>的实现类 } 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; } // Tag the request (if API >= 14) if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.ICE_CREAM_SANDWICH) { TrafficStats.setThreadStatsTag(request.getTrafficStatsTag()); } // Perform the network request. NetworkResponse networkResponse = mNetwork.performRequest(request);//2.发起网络请求 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);//3.把这个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);//4.将网络响应加入缓存 request.addMarker("network-cache-written"); } // Post the response back. request.markDelivered(); mDelivery.postResponse(request, response);//5.将网络响应发回主线程 } catch (VolleyError volleyError) { parseAndDeliverNetworkError(request, volleyError); } catch (Exception e) { VolleyLog.e(e, "Unhandled exception %s", e.toString()); mDelivery.postError(request, new VolleyError(e)); } } }
如何实现多线程下载
要理解Volley的并行性实现,必需理解PriorityBlockingQueue并发类。注意到我们在NetworkDispatcher循环中并没有使用显式的同步(使用Lock或者使用synchronize),因为PriorityBlockingQueue的实现是线程安全的。在使用显式的wait()和notifyAll()时存在的类和类之间的耦合可以因此消除,因为每个类只和BlockingQueue通信。这个知识点可以参考《Java编程思想》P713.
BlockingQueue
implementations are thread-safe. All queuing methods achieve their effects atomically using internal locks or other forms of concurrency control. 所以可以有多个NetworkDispatcher线程同时从请求队列PriorityBlockingQueue中取出请求而不会产生线程冲突,那就是Volley支持多线程下载图片的方式。
class Producer implements Runnable { private final BlockingQueue queue; Producer(BlockingQueue q) { queue = q; } public void run() { try { while (true) { queue.put(produce()); } } catch (InterruptedException ex) { ... handle ...} } Object produce() { ... } } class Consumer implements Runnable { private final BlockingQueue queue; Consumer(BlockingQueue q) { queue = q; } public void run() { try { while (true) { consume(queue.take()); } } catch (InterruptedException ex) { ... handle ...} } void consume(Object x) { ... } } class Setup { void main() { BlockingQueue q = new SomeQueueImplementation(); Producer p = new Producer(q); Consumer c1 = new Consumer(q); Consumer c2 = new Consumer(q); new Thread(p).start(); new Thread(c1).start(); new Thread(c2).start(); } }}
所以在Volley中,request就是产品,主线程是生产者,NetworkDispatcher线程是消费者。
/** 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; /** * Creates a new network dispatcher thread. You must call {@link #start()} * in order to begin processing. * * @param queue Queue of incoming requests for triage * @param network Network interface to use for performing requests * @param cache Cache interface to use for writing responses to cache * @param delivery Delivery interface to use for posting responses */ public NetworkDispatcher(BlockingQueue<Request> queue, Network network, Cache cache, ResponseDelivery delivery) { mQueue = queue; mNetwork = network; mCache = cache; mDelivery = delivery; }
NetworkDispatcher构造函数的几个参数都是接口,而run方法则使用这些策略类方法实现了算法的主体流程,具体实现有些留给了开发者,有些则是框架实现。比如ImageCache作为一级缓存的Cache方法留给了开发者实现,由开发者控制具体的缓存策略,当然Volley建议我们使用LRUCache作为L1缓存的实现。
/** * Creates the worker pool. Processing will not begin until {@link #start()} is called. * * @param cache A Cache to use for persisting responses to disk * @param network A Network interface for performing HTTP requests * @param threadPoolSize Number of network dispatcher threads to create * @param delivery A ResponseDelivery interface for posting responses and errors */ public RequestQueue(Cache cache, Network network, int threadPoolSize, ResponseDelivery delivery) { mCache = cache; mNetwork = network; mDispatchers = new NetworkDispatcher[threadPoolSize]; mDelivery = delivery; } /** * 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(); } }
如何将网络响应发回主线程
mDelivery.postResponse(request, response);
mDelivery是ResponseDelivery接口,由ExecutorDelivery实现,以下是ExecutorDelivery中postResponse的实现:
@Override public void postResponse(Request<?> request, Response<?> response) { postResponse(request, response, null); } @Override public void postResponse(Request<?> request, Response<?> response, Runnable runnable) { request.markDelivered(); request.addMarker("post-response"); mResponsePoster.execute(new ResponseDeliveryRunnable(request, response, runnable)); }
我们看看mResponsePoster.execute()是怎么实现的:
/** Used for posting responses, typically to the main thread. */ private final Executor mResponsePoster; /** * Creates a new response delivery interface. * @param handler {@link Handler} to post responses on */ public ExecutorDelivery(final Handler handler) { // Make an Executor that just wraps the handler. mResponsePoster = new Executor() { @Override public void execute(Runnable command) { handler.post(command); } }; }
没错,就是通过handler去post请求,因为要将请求发回主线程,所以这里的Handler就是主线程的Hanlder
public RequestQueue(Cache cache, Network network, int threadPoolSize) { this(cache, network, threadPoolSize, new ExecutorDelivery(new Handler(Looper.getMainLooper()))); }