/**
* 本文可以随意转载到任何网站或者App，
* BUT
* 转载也要按“基本法”，
* 请注明原文出处和作者
*/

fresco官方高大上介绍（1）（注意：前方有堵墙）

fresco官方高大上介绍（2）（注意：前方有堵墙）

介绍：

上一篇大概介绍了fresco这个lib的整体结构和流程，这篇主要介绍fresco中关键的一部分--Producer。

个人觉得，Producer基本是整个ImagePipeline module的核心，串联了整个图片读取的流程和各个细节（decode，resize等等）的处理，而且感觉整个设计上很有意思，读完感觉收益匪浅。

正文：

（分析主要是代码，相当枯燥~~~sigh）

以一次网络请求的例，进行分析，其他类型的请求，例如从cache中读取图片等，都差不多。

当SimpleDraweeView#setController后，图片拉取的流程就开始了（详细流程可以参看上一篇的流程图）。忽略掉大部分细节，流程会来到（PipelineDraweeControllerBuilder.java）：

  @Overrideprotected DataSource<CloseableReference<CloseableImage>> getDataSourceForRequest(ImageRequest imageRequest,Object callerContext,boolean bitmapCacheOnly) {if (bitmapCacheOnly) {return mImagePipeline.fetchImageFromBitmapCache(imageRequest, callerContext);} else {return mImagePipeline.fetchDecodedImage(imageRequest, callerContext);}}

这里就是开始图片拉取，转入到ImagePipeline核心流程。这里就是调用的ImagePipeline的fetchDecodeImage方法，从名字看，意思就是“获取一张decode（解码）的图片”，其代码(ImagePipeline.java)：

  /*** Submits a request for execution and returns a DataSource representing the pending decoded* image(s).* <p>The returned DataSource must be closed once the client has finished with it.* @param imageRequest the request to submit* @return a DataSource representing the pending decoded image(s)*/public DataSource<CloseableReference<CloseableImage>> fetchDecodedImage(ImageRequest imageRequest,Object callerContext) {try {Producer<CloseableReference<CloseableImage>> producerSequence =mProducerSequenceFactory.getDecodedImageProducerSequence(imageRequest);return submitFetchRequest(producerSequence,imageRequest,ImageRequest.RequestLevel.FULL_FETCH,callerContext);} catch (Exception exception) {return DataSources.immediateFailedDataSource(exception);}}

这里有两个主要的函数：getDecodedImageProducerSequence和submitFetchRequest

getDecodedImageProducerSequence的返回值就是一个Producer<CloseableReference<CloseableImage>>。该值作为参数，传到第二个函数submitFetchRequest中，先看submitFetchRequest：

  private <T> DataSource<CloseableReference<T>> submitFetchRequest(Producer<CloseableReference<T>> producerSequence,ImageRequest imageRequest,ImageRequest.RequestLevel lowestPermittedRequestLevelOnSubmit,Object callerContext) {try {ImageRequest.RequestLevel lowestPermittedRequestLevel =ImageRequest.RequestLevel.getMax(imageRequest.getLowestPermittedRequestLevel(),lowestPermittedRequestLevelOnSubmit);SettableProducerContext settableProducerContext = new SettableProducerContext(imageRequest,generateUniqueFutureId(),mRequestListener,callerContext,lowestPermittedRequestLevel,
/* isPrefetch */ false,imageRequest.getProgressiveRenderingEnabled() ||!UriUtil.isNetworkUri(imageRequest.getSourceUri()),imageRequest.getPriority());return CloseableProducerToDataSourceAdapter.create(producerSequence,settableProducerContext,mRequestListener);} catch (Exception exception) {return DataSources.immediateFailedDataSource(exception);}}

然后上面的create函数，就会一直到（AbstractProducerToDataSourceAdapter.java）：

  protected AbstractProducerToDataSourceAdapter(Producer<T> producer,SettableProducerContext settableProducerContext,RequestListener requestListener) {mSettableProducerContext = settableProducerContext;mRequestListener = requestListener;mRequestListener.onRequestStart(settableProducerContext.getImageRequest(),mSettableProducerContext.getCallerContext(),mSettableProducerContext.getId(),mSettableProducerContext.isPrefetch());producer.produceResults(createConsumer(), settableProducerContext);}

到最后，submitFetchRequest会调用到Producer#produceResults方法，而这个producer就是前面那个getDecodedImageProducerSequence方法产生的，所以回头看这个最最关键的地方。

getDecodedImageProducerSequence是ProducerSequenceFactory.java的方法：

带着疑问继续看：

private Producer<CloseableReference<CloseableImage>> getBasicDecodedImageSequence(ImageRequest imageRequest) {Preconditions.checkNotNull(imageRequest);Uri uri = imageRequest.getSourceUri();Preconditions.checkNotNull(uri, "Uri is null.");if (UriUtil.isNetworkUri(uri)) {return getNetworkFetchSequence();} else if (UriUtil.isLocalFileUri(uri)) {if (MediaUtils.isVideo(MediaUtils.extractMime(uri.getPath()))) {return getLocalVideoFileFetchSequence();} else {return getLocalImageFileFetchSequence();}} else if (UriUtil.isLocalContentUri(uri)) {return getLocalContentUriFetchSequence();} else if (UriUtil.isLocalAssetUri(uri)) {return getLocalAssetFetchSequence();} else if (UriUtil.isLocalResourceUri(uri)) {return getLocalResourceFetchSequence();} else if (UriUtil.isDataUri(uri)) {return getDataFetchSequence();} else {String uriString = uri.toString();if (uriString.length() > 30) {uriString = uriString.substring(0, 30) + "...";}throw new RuntimeException("Unsupported uri scheme! Uri is: " + uriString);}}

看代码可知道，就是根据ImageRequest的Uri，选择一个sequence Producer，我们这里假设是网络请求图片，所以选择的是图中红色方法getNetworkFetchSequence，它也是返回一个Producer（可粗略看）：

  /*** swallow result if prefetch -> bitmap cache get ->* background thread hand-off -> multiplex -> bitmap cache -> decode -> multiplex ->* encoded cache -> disk cache -> (webp transcode) -> network fetch.*/private synchronized Producer<CloseableReference<CloseableImage>> getNetworkFetchSequence() {if (mNetworkFetchSequence == null) {mNetworkFetchSequence =newBitmapCacheGetToDecodeSequence(getCommonNetworkFetchToEncodedMemorySequence());}return mNetworkFetchSequence;}

先看红色代码，getCommonNetworkFetchToEncodedMemorySequence，它也是返回一个Producer（可粗略看）：

  /*** multiplex -> encoded cache -> disk cache -> (webp transcode) -> network fetch.*/private synchronized Producer<EncodedImage> getCommonNetworkFetchToEncodedMemorySequence() {if (mCommonNetworkFetchToEncodedMemorySequence == null) {Producer<EncodedImage> inputProducer =newEncodedCacheMultiplexToTranscodeSequence(mProducerFactory.newNetworkFetchProducer(mNetworkFetcher));mCommonNetworkFetchToEncodedMemorySequence =ProducerFactory.newAddImageTransformMetaDataProducer(inputProducer);if (mResizeAndRotateEnabledForNetwork && !mDownsampleEnabled) {mCommonNetworkFetchToEncodedMemorySequence =mProducerFactory.newResizeAndRotateProducer(mCommonNetworkFetchToEncodedMemorySequence);}}return mCommonNetworkFetchToEncodedMemorySequence;}

再看，newEncodedCacheMultiplexToTranscodeSequence，它也是返回一个Producer（可粗略看）：

  /*** encoded cache multiplex -> encoded cache -> (disk cache) -> (webp transcode)* @param inputProducer producer providing the input to the transcode* @return encoded cache multiplex to webp transcode sequence*/private Producer<EncodedImage> newEncodedCacheMultiplexToTranscodeSequence(Producer<EncodedImage> inputProducer) {if (Build.VERSION.SDK_INT < Build.VERSION_CODES.JELLY_BEAN_MR2) {inputProducer = mProducerFactory.newWebpTranscodeProducer(inputProducer);}inputProducer = mProducerFactory.newDiskCacheProducer(inputProducer);EncodedMemoryCacheProducer encodedMemoryCacheProducer =mProducerFactory.newEncodedMemoryCacheProducer(inputProducer);return mProducerFactory.newEncodedCacheKeyMultiplexProducer(encodedMemoryCacheProducer);}

选其中一个红色函数看，newDiskCacheProducer，它也是返回一个Producer（可粗略看）：

  public DiskCacheProducer newDiskCacheProducer(Producer<EncodedImage> inputProducer) {return new DiskCacheProducer(mDefaultBufferedDiskCache,mSmallImageBufferedDiskCache,mCacheKeyFactory,inputProducer);}

好了，到此为止（列出的函数足够多了，晕~~~）

其实上面，一连串几个函数，如果有细心的留意，它们有一个特点，就是，（每一个函数）都以 （上一个函数）产生的Producer作为参数进行传递。

很熟悉了吧，可以把sequence Producer就看成是上面这样的一个逻辑~~~

那么这样做的作用是什么？我们选一个简单Producer来分析，就以DiskCacheProducer为例（留意代码绿色注释的描述）：

 public DiskCacheProducer(BufferedDiskCache defaultBufferedDiskCache,BufferedDiskCache smallImageBufferedDiskCache,CacheKeyFactory cacheKeyFactory,Producer<EncodedImage> inputProducer) {mDefaultBufferedDiskCache = defaultBufferedDiskCache;mSmallImageBufferedDiskCache = smallImageBufferedDiskCache;mCacheKeyFactory = cacheKeyFactory;mInputProducer = inputProducer;}public void produceResults(final Consumer<EncodedImage> consumer,final ProducerContext producerContext) {ImageRequest imageRequest = producerContext.getImageRequest();
//如果diskcache disabled的话，那么直接执行maybeStartInputProducerif (!imageRequest.isDiskCacheEnabled()) {maybeStartInputProducer(consumer, consumer, producerContext);return;}final ProducerListener listener = producerContext.getListener();final String requestId = producerContext.getId();listener.onProducerStart(requestId, PRODUCER_NAME);final CacheKey cacheKey = mCacheKeyFactory.getEncodedCacheKey(imageRequest);final BufferedDiskCache cache =imageRequest.getImageType() == ImageRequest.ImageType.SMALL? mSmallImageBufferedDiskCache: mDefaultBufferedDiskCache;Continuation<EncodedImage, Void> continuation = new Continuation<EncodedImage, Void>() {//回调
      @Overridepublic Void then(Task<EncodedImage> task)throws Exception {//根据task是canceled，fault等状态决定如何执行if (task.isCancelled() ||(task.isFaulted() && task.getError() instanceof CancellationException)) {listener.onProducerFinishWithCancellation(requestId, PRODUCER_NAME, null);consumer.onCancellation();} else if (task.isFaulted()) {listener.onProducerFinishWithFailure(requestId, PRODUCER_NAME, task.getError(), null);//出错了，就调用maybeStartInputProducer
          maybeStartInputProducer(consumer,new DiskCacheConsumer(consumer, cache, cacheKey),producerContext);} else {EncodedImage cachedReference = task.getResult();if (cachedReference != null) {listener.onProducerFinishWithSuccess(requestId,PRODUCER_NAME,getExtraMap(listener, requestId, true));consumer.onProgressUpdate(1);consumer.onNewResult(cachedReference, true);cachedReference.close();} else {//没有结果，就调用maybeStartInputProducer
            listener.onProducerFinishWithSuccess(requestId,PRODUCER_NAME,getExtraMap(listener, requestId, false));maybeStartInputProducer(consumer,new DiskCacheConsumer(consumer, cache, cacheKey),producerContext);}}return null;}};AtomicBoolean isCancelled = new AtomicBoolean(false);final Task<EncodedImage> diskCacheLookupTask =cache.get(cacheKey, isCancelled);
//执行task，task其实就是从缓存中取结果，执行后，前面的continuation就会被回调
    diskCacheLookupTask.continueWith(continuation);subscribeTaskForRequestCancellation(isCancelled, producerContext);}//调用mInputProducer的produceResults private void maybeStartInputProducer(Consumer<EncodedImage> consumerOfDiskCacheProducer,Consumer<EncodedImage> consumerOfInputProducer,ProducerContext producerContext) {if (producerContext.getLowestPermittedRequestLevel().getValue() >=ImageRequest.RequestLevel.DISK_CACHE.getValue()) {consumerOfDiskCacheProducer.onNewResult(null, true);return;}mInputProducer.produceResults(consumerOfInputProducer, producerContext);}

从前面知道，当开始拉取图片的时候，Producer#produceResult开始执行，注释标出了关键的步骤，从这些步骤可以看出，其实DiskCacheProducer拉取图片时，做的任务大概就是：先看Diskcache中是否有缓存的图片，如果有，就直接返回缓存，如果没有，就用inputProducer来处理。

然后，inputProducer处理完结果会怎样呢？它处理的结果会在consumer中接收到，上面的例子代码对应的就是DiskCacheConsumer（留意绿色注释）：

  /*** Consumer that consumes results from next producer in the sequence.** <p>The consumer puts the last result received into disk cache, and passes all results (success* or failure) down to the next consumer.*/private class DiskCacheConsumer extends DelegatingConsumer<EncodedImage, EncodedImage> {private final BufferedDiskCache mCache;private final CacheKey mCacheKey;private DiskCacheConsumer(final Consumer<EncodedImage> consumer,final BufferedDiskCache cache,final CacheKey cacheKey) {super(consumer);mCache = cache;mCacheKey = cacheKey;}//inputProducer的结果会从这里返回，即newResult
    @Overridepublic void onNewResultImpl(EncodedImage newResult, boolean isLast) {//返回的结果加入cache中if (newResult != null && isLast) {mCache.put(mCacheKey, newResult);}//回调上一层procducer传进来的consumer
      getConsumer().onNewResult(newResult, isLast);}}

到这里，应该就大概明白这个sequence Producer的作用了，所谓sequence Producer，其实就是一层层的Producer不断的嵌套连接起来，完成同一个任务，而每一个Producer都相互独立，完成各自任务；同时，Producer间产生的结果，也会相互传递，互为表里。可以称其为“Producer链”，但是“Producer链”本身被抽象成一个Producer，那么对于上层来看，这样一个复杂的处理逻辑就被隐藏起来了，变得更加容易理解。

sequence Producer功能极其强大，不同的Producer的组合，产生了很多不同的效果，对于代码的扩展性，可复用性和灵活性都有很大好处。

例如，MultiplexProducer：

注释：

/**
* Producer for combining multiple identical requests into a single request.
*
* <p>Requests using the same key will be combined into a single request. This request is only
* cancelled when all underlying requests are cancelled, and returns values to all underlying
* consumers. If the request has already return one or more results but has not finished, then
* any requests with the same key will have the most recent result returned to them immediately.
*
* @param <K> type of the key
* @param <T> type of the closeable reference result that is returned to this producer
*/
@ThreadSafe
public abstract class MultiplexProducer<K, T extends Closeable> implements Producer<T>

理解为，多路复用Producer（什么鬼东西？），其实就是将相同的任务合并为一个，例如相同url的重复请求，如何做到的，关键代码（主要看注释）：

  @Overridepublic void produceResults(Consumer<T> consumer, ProducerContext context) {K key = getKey(context);Multiplexer multiplexer;boolean createdNewMultiplexer;
// We do want to limit scope of this lock to guard only accesses to mMultiplexers map.
// However what we would like to do here is to atomically lookup mMultiplexers, add new
// consumer to consumers set associated with the map's entry and call consumer's callback with
// last intermediate result. We should not do all of those things under this lock.do {createdNewMultiplexer = false;synchronized (this) {//根据key获得多路复用器，当缓存没有的时候，才create一个，不然直接忽略multiplexer = getExistingMultiplexer(key);if (multiplexer == null) {multiplexer = createAndPutNewMultiplexer(key);createdNewMultiplexer = true;}}
// addNewConsumer may call consumer's onNewResult method immediately. For this reason
// we release "this" lock. If multiplexer is removed from mMultiplexers in the meantime,
// which is not very probable, then addNewConsumer will fail and we will be able to retry.} while (!multiplexer.addNewConsumer(consumer, context));//如果前面没有创建，也就是存在缓存的多路复用器，那么就不会调用startInputProducerIfHasAttachedConsumers，然后inputProducer就不起作用了，这样，就起到合并请求的作用if (createdNewMultiplexer) {multiplexer.startInputProducerIfHasAttachedConsumers();}}

不同的功能Producer还有很多，例如对图片进行resize和rotate的ResizeAndRotateProducer， 异步执行任务的ThreadHandoffProducer等等，如此灵活的实现，得益于sequence Factory这种设计。

总结

ImagePipeline的核心Producer，通过sequence的形式，很好的串联了整个图片网络读取，缓存，bitmap处理等流程，通过优秀的设计，保证的代码高扩展性，高可复用性和高灵活性。

这种设计刚好对应就是“pipeline”这个词的含义，就如现代的pipelined CPU一样，把对指令的处理，拆分成多个stage，同时输入输出相互依赖和协作，共同完成一个任务。

~~~文卒~~~