Fresco源码分析四

Producer相关源码在com.facebook.imagepipeline.producers包内，类和接口加在一起大概40多个。
先看Producer接口的代码：

public interface Producer<T> {
void produceResults(Consumer<T> consumer, ProducerContext context);
}

Producer：Producer是在image pipeline中用于图片处理模块。
处理一个图片请求包含不同的任务，例如从网上获取，硬盘缓存，内存缓存，解码，转换等等。Producer 代表了一个单独的任务，其执行结果是T的对象。这就把一个完整的请求拆分成了一系列的Producer，Producer允许我们构建不同请求时复用这些模块。

ProducerContext：被用于传递上下文信息给Producer；
它的指责是通知Producer哪张图片该进行fetched/decoded/resized/cached etc的处理。

ProducerContextCallbacks：request状态改变回调，如request被取消，是否预加载，是否获取中间结果，request优先级的改变。

ProducerListener：Producer的监听类。回调方法有Producer任务开始，Producer发布事件，Producer成功结束，Producer失败结束，Producer取消结束等。

Consumer<T>：消费者，消费Producer返回的信息。

public interface Consumer<T> {
  void onNewResult(T newResult, boolean isLast);
  void onFailure(Throwable t);
  void onCancellation();
  void onProgressUpdate(float progress);
}

BaseConsumer Consumer的基本实现类
DelegatingConsumer Consumer的委托类
获取一系列的Producer，这些Producer是层层包含关系：

Producer<CloseableReference<CloseableImage>> producerSequence =
          mProducerSequenceFactory.getDecodedImageProducerSequence(imageRequest);

getDecodedImageProducerSequence()方法内部：

Producer<CloseableReference<CloseableImage>> pipelineSequence =
        getBasicDecodedImageSequence(imageRequest);

getBasicDecodedImageSequence()方法内部通过ImageRequest的getSourceUri()方法获取到图片的Uri,根据不同的Uri返回不同类型的Producer。
比如是HTTP和HTTPS开头的Uri，调用getNetworkFetchSequence()。
这个函数的请求链基本是下面的顺序：
swallow result if prefetch ->
bitmap cache get ->
background thread hand-off ->
multiplex ->
bitmap cache ->
decode ->
multiplex ->
encoded cache ->
disk cache ->
(webp transcode) ->
network fetch
fresco获取图片就是不断的把自己定义为一个consumer，然后向相应的producer请求数据(produceResults)的过程。（参考自该文章）
形象地看，类似是一个Producer吞（swallow ）下了另一个Producer。
NetworkFetchProducer
WebpTranscodeProducer
DiskCacheProducer
EncodedMemoryCacheProducer
EncodedCacheKeyMultiplexProducer
AddImageTransformMetaDataProducer
ResizeAndRotateProducer
BitmapMemoryCacheProducer
BitmapMemoryCacheKeyMultiplexProducer
BackgroundThreadHandoffProducer
BitmapMemoryCacheGetProducer
在这一请求流水总共用到了以上的Producer。从上到下，上一层的Producer是一下层的Producer的构造函数中的inputProducer。
下面引入一段官方文档中”不用担心，使用streaming吧”这一小节：
过去，Java上的异步代码常常通过类似Future的机制来执行。代码被提交给其他线程，同时检查一个类似Future的对象，看是否能得出结果。然而，假设只会有一个结果。当处理渐进式图像的时候，我们希望有一个完整而持续的结果。

我们的解决方案是使用Future的更广义的版本，叫做DataSource。它提供了一个订阅方法，其调用者一定要传递一个DataSubscriber和一个Executor。该DataSubscriber从DataSource的中间值和最终结果那里接收通知，并且提供一个简单的方法来区分两者。因为我们常常处理对象，这就需要一个明确的close调用，DataSource本身就实现了Closeable。

深入分析之后，其实上述的每一个格子用一个新的框架来实现，叫做Producer/Consumer。绘制图示的灵感来源于ReactiveX框架。我们的系统和RxJava有相似的接口，但其更适于移动设备，并内建了Closeables支持。

该接口保持简单。Producer仅有一个方法produceResults，其中包含了一个Consumer对象。Consumer反过来也有有一个onNewResult方法。

我们用此系统组成了一个生产链。设想我们有这样的一个生产者，其工作是转换 I 类型到 O 类型。实现如下：

public class OutputProducer<I, O> implements Producer<O> {
 
	private final Producer<I> mInputProducer;
 
	public OutputProducer(Producer<I> inputProducer) {
		this.mInputProducer = inputProducer;
	}
 
	public void produceResults(Consumer<O> outputConsumer, ProducerContext context) {
		Consumer<I> inputConsumer = new InputConsumer(outputConsumer);
		mInputProducer.produceResults(inputConsumer, context);
	}
 
	private static class InputConsumer implements Consumer<I> {
		private final Consumer<O> mOutputConsumer;
		public InputConsumer(Consumer<O> outputConsumer) {
			mOutputConsumer = outputConsumer;
		} 
		public void onNewResult(I newResult, boolean isLast) {
			O output = doActualWork(newResult);
			mOutputConsumer.onNewResult(output, isLast); 
		}
	}
}

所以Producer-Consumer的关系就是这样的，全部Producer内部的代码都是这样的逻辑。
比如BitmapMemoryCacheProducer，produceResults()方法内部的关键代码如下：

final ImageRequest imageRequest = producerContext.getImageRequest();
final CacheKey cacheKey = mCacheKeyFactory.getBitmapCacheKey(imageRequest); 
CloseableReference<CloseableImage> cachedReference = mMemoryCache.get(cacheKey);
  if (cachedReference != null) {
      boolean isFinal = cachedReference.get().getQualityInfo().isOfFullQuality();
      if (isFinal) {
        listener.onProducerFinishWithSuccess(
            requestId,
            getProducerName(),
            listener.requiresExtraMap(requestId) ? ImmutableMap.of(VALUE_FOUND, "true") : null);
        consumer.onProgressUpdate(1f);
      }
      consumer.onNewResult(cachedReference, isFinal);
      cachedReference.close();
      if (isFinal) {
        return;
      }
    }

如从mMemoryCache中查找到图片，就调用consumer.onNewResult()，然后返回。
如果没有查找到，就执行如下代码:

Consumer<CloseableReference<CloseableImage>> wrappedConsumer = wrapConsumer(consumer, cacheKey);
mInputProducer.produceResults(wrappedConsumer, producerContext);

调用wrapConsumer()方法对原始的consumer进行包装，然后调用mInputProducer（在这里是BackgroundThreadHandoffProducer）的produceResults()方法。
BitmapMemoryCacheProducer的wrapConsumer()方法执行了如下处理：
1）数据为空，直接传给上层Consumer处理并返回；

if (newResult == null) {
          if (isLast) {
            getConsumer().onNewResult(null, true);
          }
          return;
        }

2）数据存储着某种状态（如动态图存储着当前浏览帧）时，不进行缓存，直接传给上层Consumer处理并返回；

if (newResult.get().isStateful()) {
         getConsumer().onNewResult(newResult, isLast);
         return;
       }

3）当没有结束传递时，如果收到数据的质量大于缓存中对应数据的质量（如果存在的话）时，则传给上层Consumer处理并返回；

if (!isLast) {
          CloseableReference<CloseableImage> currentCachedResult = mMemoryCache.get(cacheKey);
          if (currentCachedResult != null) {
            try {
              QualityInfo newInfo = newResult.get().getQualityInfo();
              QualityInfo cachedInfo = currentCachedResult.get().getQualityInfo();
              if (cachedInfo.isOfFullQuality() || cachedInfo.getQuality() >= newInfo.getQuality()) {
                getConsumer().onNewResult(currentCachedResult, false);
                return;
              }
            } finally {
              CloseableReference.closeSafely(currentCachedResult);
            }
          }
        }
        // cache and forward the new result
        CloseableReference<CloseableImage> newCachedResult =
            mMemoryCache.cache(cacheKey, newResult);

4）数据传递结束，将得到的数据缓存起来，更新进度，通知上层Consumer处理。

try {
     if (isLast) {
            getConsumer().onProgressUpdate(1f);
          }
          getConsumer().onNewResult(
              (newCachedResult != null) ? newCachedResult : newResult, isLast);
        } finally {
          CloseableReference.closeSafely(newCachedResult);
        }

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);
    }
  }

SettableProducerContext继承自BaseProducerContext，并可设置是否预处理，是否获取中间结果，request的优先级。

CloseableProducerToDataSourceAdapter<T>extends AbstractProducerToDataSourceAdapter<CloseableReference<T>

CloseableProducerToDataSourceAdapter：
DataSource> backed by a Producer>

AbstractProducerToDataSourceAdapter<T> extends AbstractDataSource<T>

AbstractProducerToDataSourceAdapter：
DataSource backed by a Producer

AbstractProducerToDataSourceAdapter先通过createConsumer()方法创建一个Consumer，在onNewResultImpl，onFailureImpl，onCancellationImpl，onProgressUpdateImpl中调用AbstractProducerToDataSourceAdapter自己的实现。然后调用producer.produceResults开始加载数据。

参考文章：https://github.com/desmond1121/Fresco-Source-Analysis