Android-Universal-Image-Loader 学习笔记（五）线程池分析

                         UniveralImageLoader中的线程池

           一般情况网络访问就需要App创建一个线程来执行（不然可能出现很臭的ANR），但是这也导致了当网络访问比较多的情况下，线程的数目可能指数增多，虽然Android系统理论上说可以创建无数个线程,但是某一时间段，线程数的急剧增加可能导致系统OOM。

       在UIL中引入了线程池这种技术来管理线程。合理利用线程池能够带来三个好处。

       第一：降低资源消耗。通过重复利用已创建的线程降低线程创建和销毁造成的消耗。

       第二：提高响应速度。当任务到达时，任务可以不需要等到线程创建就能立即执行。

       第三：提高线程的可管理性。线程是稀缺资源，如果无限制的创建，不仅会消耗系统资源，还会降低系统的稳定性，使用线程池可以进行统一的分配，调优和监控。

 

        当需要加载显示图片的时候，相关的task通过engine.submit(...)函数提交执行，那么submit之后发生了什么呢？engine是ImageLoaderEngine类的一个实例，他主要用来响应displayTask的执行。我们跟进ImageLoaderEngine中看看相关的字段和方法。

class ImageLoaderEngine {

    final ImageLoaderConfiguration configuration;

    private Executor taskExecutor;
    private Executor taskExecutorForCachedImages;
    private Executor taskDistributor;

    private final Map<Integer, String> cacheKeysForImageAwares = Collections
            .synchronizedMap(new HashMap<Integer, String>());
    private final Map<String, ReentrantLock> uriLocks = new WeakHashMap<String, ReentrantLock>();

    private final AtomicBoolean paused = new AtomicBoolean(false);
    private final AtomicBoolean networkDenied = new AtomicBoolean(false);
    private final AtomicBoolean slowNetwork = new AtomicBoolean(false);

    private final Object pauseLock = new Object();

    ImageLoaderEngine(ImageLoaderConfiguration configuration) {
        this.configuration = configuration;

        taskExecutor = configuration.taskExecutor;
        taskExecutorForCachedImages = configuration.taskExecutorForCachedImages;

        taskDistributor = DefaultConfigurationFactory.createTaskDistributor();
    }

    /** Submits task to execution pool */
    void submit(final LoadAndDisplayImageTask task) {
        taskDistributor.execute(new Runnable() {
            @Override
            public void run() {
                File image = configuration.diskCache.get(task.getLoadingUri());
                boolean isImageCachedOnDisk = image != null && image.exists();
                initExecutorsIfNeed();
                if (isImageCachedOnDisk) {
                    taskExecutorForCachedImages.execute(task);
                } else {
                    taskExecutor.execute(task);
                }
            }
        });
    }

    /** Submits task to execution pool */
    void submit(ProcessAndDisplayImageTask task) {
        initExecutorsIfNeed();
        taskExecutorForCachedImages.execute(task);
    }

    private void initExecutorsIfNeed() {
        if (!configuration.customExecutor && ((ExecutorService) taskExecutor).isShutdown()) {
            taskExecutor = createTaskExecutor();
        }
        if (!configuration.customExecutorForCachedImages && ((ExecutorService) taskExecutorForCachedImages)
                .isShutdown()) {
            taskExecutorForCachedImages = createTaskExecutor();
        }
    }

    private Executor createTaskExecutor() {
        return DefaultConfigurationFactory
                .createExecutor(configuration.threadPoolSize, configuration.threadPriority,
                configuration.tasksProcessingType);
    }

    /**
     * Returns URI of image which is loading at this moment into passed {@link com.nostra13.universalimageloader.core.imageaware.ImageAware}
     */
    String getLoadingUriForView(ImageAware imageAware) {
        return cacheKeysForImageAwares.get(imageAware.getId());
    }

    /**
     * Associates <b>memoryCacheKey</b> with <b>imageAware</b>. Then it helps to define image URI is loaded into View at
     * exact moment.
     */
    void prepareDisplayTaskFor(ImageAware imageAware, String memoryCacheKey) {
        cacheKeysForImageAwares.put(imageAware.getId(), memoryCacheKey);
    }

    /**
     * Cancels the task of loading and displaying image for incoming <b>imageAware</b>.
     *
     * @param imageAware {@link com.nostra13.universalimageloader.core.imageaware.ImageAware} for which display task
     *                   will be cancelled
     */
    void cancelDisplayTaskFor(ImageAware imageAware) {
        cacheKeysForImageAwares.remove(imageAware.getId());
    }

    /**
     * Denies or allows engine to download images from the network.<br /> <br /> If downloads are denied and if image
     * isn't cached then {@link ImageLoadingListener#onLoadingFailed(String, View, FailReason)} callback will be fired
     * with {@link FailReason.FailType#NETWORK_DENIED}
     *
     * @param denyNetworkDownloads pass <b>true</b> - to deny engine to download images from the network; <b>false</b> -
     *                             to allow engine to download images from network.
     */
    void denyNetworkDownloads(boolean denyNetworkDownloads) {
        networkDenied.set(denyNetworkDownloads);
    }

    /**
     * Sets option whether ImageLoader will use {@link FlushedInputStream} for network downloads to handle <a
     * href="http://code.google.com/p/android/issues/detail?id=6066">this known problem</a> or not.
     *
     * @param handleSlowNetwork pass <b>true</b> - to use {@link FlushedInputStream} for network downloads; <b>false</b>
     *                          - otherwise.
     */
    void handleSlowNetwork(boolean handleSlowNetwork) {
        slowNetwork.set(handleSlowNetwork);
    }

    /**
     * Pauses engine. All new "load&display" tasks won't be executed until ImageLoader is {@link #resume() resumed}.<br
     * /> Already running tasks are not paused.
     */
    void pause() {
        paused.set(true);
    }

    /** Resumes engine work. Paused "load&display" tasks will continue its work. */
    void resume() {
        paused.set(false);
        synchronized (pauseLock) {
            pauseLock.notifyAll();
        }
    }

    /**
     * Stops engine, cancels all running and scheduled display image tasks. Clears internal data.
     * <br />
     * <b>NOTE:</b> This method doesn't shutdown
     * {@linkplain com.nostra13.universalimageloader.core.ImageLoaderConfiguration.Builder#taskExecutor(java.util.concurrent.Executor)
     * custom task executors} if you set them.
     */
    void stop() {
        if (!configuration.customExecutor) {
            ((ExecutorService) taskExecutor).shutdownNow();
        }
        if (!configuration.customExecutorForCachedImages) {
            ((ExecutorService) taskExecutorForCachedImages).shutdownNow();
        }

        cacheKeysForImageAwares.clear();
        uriLocks.clear();
    }

    void fireCallback(Runnable r) {
        taskDistributor.execute(r);
    }

    ReentrantLock getLockForUri(String uri) {
        ReentrantLock lock = uriLocks.get(uri);
        if (lock == null) {
            lock = new ReentrantLock();
            uriLocks.put(uri, lock);
        }
        return lock;
    }

    AtomicBoolean getPause() {
        return paused;
    }

    Object getPauseLock() {
        return pauseLock;
    }

    boolean isNetworkDenied() {
        return networkDenied.get();
    }

    boolean isSlowNetwork() {
        return slowNetwork.get();
    }
}

           从代码来看，通过taskDistributor.execute来执行一个Runnable对象的run()，从代码中不难知道它就是先试读取磁盘缓存，再根据isImageCachedOnDisk判断文件是否有缓存在磁盘中，最后通过不同的taskExecutor来执行对应的任务。我们注意到这个submit函数中出现了taskExecutorForCachedImages、taskExecutor、taskDistributor这三个对象。

           

private Executor taskExecutor;
private Executor taskExecutorForCachedImages;
private Executor taskDistributor;

/** Submits task to execution pool */
void submit(final LoadAndDisplayImageTask task) {
    taskDistributor.execute(new Runnable() {
        @Override
        public void run() {
            File image = configuration.diskCache.get(task.getLoadingUri());
            boolean isImageCachedOnDisk = image != null && image.exists();
            initExecutorsIfNeed();
            if (isImageCachedOnDisk) {
                taskExecutorForCachedImages.execute(task);
            } else {
                taskExecutor.execute(task);
            }
        }
    });
}

           

public interface Executor {
    void execute(Runnable var1);
}

 

        三者都是 Executor接口的实例。

        Executor：

Executor接口执行已提交的 Runnable 任务的对象。此接口提供一种将任务提交与每个任务将如何运行的机制（包括线程使用的细节、调度等）分离开来的方法。通常使用 Executor 而不是显式地创建线程。例如，可能会使用以下方法，而不是为一组任务中的每个任务调用 new          Thread(new(RunnableTask())).start()：

       Executor executor = anExecutor;
       executor.execute(new RunnableTask1());
       executor.execute(new RunnableTask2());
       ...

      通过Executor 对象将线程放入线程池中运行的。

      taskExecutorForCachedImages、taskExecutor

       在DefaultConfigurationFactory.createExecutor中被初始化成ThreadPoolExecutor类型的对象（这是默认情况）。需要注意的是，ThreadPoolExecutor其实是实现了ExecutorService接口的一个实体类。线程池实际表现为 ExecutorService 类的一個实例。通过使用 ExecutorService ，我们可以提交将在未来完成的任务。需要补充说明的是，ExecutorService继承自Executor接口。

      

if (taskExecutor == null) {
                taskExecutor = DefaultConfigurationFactory
                        .createExecutor(threadPoolSize, threadPriority, tasksProcessingType);
            } else {
                customExecutor = true;
            }
            if (taskExecutorForCachedImages == null) {
                taskExecutorForCachedImages = DefaultConfigurationFactory
                        .createExecutor(threadPoolSize, threadPriority, tasksProcessingType);
            } else {
                customExecutorForCachedImages = true;
            }

public static Executor createExecutor(int threadPoolSize, int threadPriority,
            QueueProcessingType tasksProcessingType) {
        boolean lifo = tasksProcessingType == QueueProcessingType.LIFO;
        BlockingQueue<Runnable> taskQueue =
                lifo ? new LIFOLinkedBlockingDeque<Runnable>() : new LinkedBlockingQueue<Runnable>();
        return new ThreadPoolExecutor(threadPoolSize, threadPoolSize, 0L, TimeUnit.MILLISECONDS, taskQueue,
                createThreadFactory(threadPriority, "uil-pool-"));
    }

方法如下：

public ThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime,

TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler)

创建一个ThreadPoolExecutor需要的参数：

• corePoolSize（线程池的基本大小）：
•         当提交一个任务到线程池时，线程池会创建一个线程来执行任务，即使其他空闲的基本线程能够执行新任务也会创建线程，等到需要执行的任务数大于线程池基本大小时就不再创建。如果调用了线程池的prestartAllCoreThreads方法，线程池会提前创建并启动所有基本线程。
• runnableTaskQueue（任务队列）：
•        用于保存等待执行的任务的阻塞队列。 可以选择以下几个阻塞队列。
  • 1）ArrayBlockingQueue：
  •        是一个基于数组结构的有界阻塞队列，此队列按 FIFO（先进先出）原则对元素进行排序。
  • 2）LinkedBlockingQueue：
  •        一个基于链表结构的阻塞队列，此队列按FIFO （先进先出） 排序元素，吞吐量通常要高于                                                                     ArrayBlockingQueue。静态工厂方法Executors.newFixedThreadPool()使用了这个队列。
  • 3）SynchronousQueue：
  •        一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态，吞吐量通常要高于LinkedBlockingQueue，静态工厂方法Executors.newCachedThreadPool使用了这个队列。
  • 4）PriorityBlockingQueue：一个具有优先级的无限阻塞队列。
• maximumPoolSize（线程池最大大小）：
•        线程池允许创建的最大线程数。如果队列满了，并且已创建的线程数小于最大线程数，则线程池会再创建新的线程执行任务。值得注意的是如果使用了无界的任务队列这个参数就没什么效果。
• ThreadFactory：
•       用于设置创建线程的工厂，可以通过线程工厂给每个创建出来的线程设置更有意义的名字。
• RejectedExecutionHandler（饱和策略）：
•       当队列和线程池都满了，说明线程池处于饱和状态，那么必须采取一种策略处理提交的新任务。这个策略默认情况下是AbortPolicy，表示无法处理新任务时抛出异常。以下是JDK1.5提供的四种策略。
  • 1）AbortPolicy：直接抛出异常。
  • 2）CallerRunsPolicy：只用调用者所在线程来运行任务。
  • 3）DiscardOldestPolicy：丢弃队列里最近的一个任务，并执行当前任务。
  • 4）DiscardPolicy：不处理，丢弃掉。
  • 当然也可以根据应用场景需要来实现RejectedExecutionHandler接口自定义策略。如记录日志或持久化不能处理的任务。
• keepAliveTime（线程活动保持时间）：
•       线程池的工作线程空闲后，保持存活的时间。所以如果任务很多，并且每个任务执行的时间比较短，可以调大这个时间，提高线程的利用率。
• TimeUnit（线程活动保持时间的单位）：
•        可选的单位有天（DAYS），小时（HOURS），分钟（MINUTES），毫秒(MILLISECONDS)，微秒(MICROSECONDS, 千分之一毫秒)和毫微秒(NANOSECONDS, 千分之一微秒)。

      taskDistributor

 

接下来，再让我们分析taskDistributor的创建过程。分析发现，taskDistributor在DefaultConfigurationFactory.createTaskDistributor()中被创建，它是一个ThreadPoolExecutor类型的对象，通过Executors线程池工厂创建。官方文档推荐程序员用它来创建线程池，因为它已经配置好常见的线程池情景。接下来让我们来了解一下Executors工厂方法所能创建的线程池类型。

用Executors静态工厂方法创建的线程池类型：

    a) newFixedThreadPool：创建一个定长的线程池。达到最大线程数后，线程数不再增长。如果一个线程由于非预期Exception而结束，线程池会补充一个新的线程。

    b) newCachedThreadPool：创建一个可缓存的线程池。当池长度超过处理需求时，可以回收空闲的线程。

    c) newSingleThreadPool：创建一个单线程executor。

    d) newScheduledThreadPool：创建一个定长的线程池，而且支持定时的以及周期性的任务执行。类似于Timer。但是，Timer是基于绝对时间，对系统时钟的改变是敏感的，而ScheduledThreadPoolExecutor只支持相对时间。

         1) Timer是创建唯一的线程来执行所有的timer任务。如果一个任务超时了，会导致其他的TimerTask时间准确性出问题。

         2）如果TimerTask抛出uncheck 异常，Timer将会产生无法预料的行为。因此，ScheduledThreadPoolExecutor可以完全代替Timer。

 

从函数中分析可以得知：taskDistributor用来尝试读取磁盘中是否有图片缓存，因为涉及磁盘操作，需要用线程来执行。根据是否有对应的图片缓存，将图片加载的任务分发到对应的执行器。如果图片已经缓存在磁盘，则通过taskExecutorForCachedImages执行，如果图片没有缓存在磁盘,则通过taskExecutor执行。我们注意到这三个都实现了Executor接口，那么为什么要将任务细分在三个线程池中进行呢？这其实这跟线程池的调优有关，如果我们将所有的任务都放在同一个线程池中运行当然是可以的，但是这样的话所有的任务就都只能采取同一种任务优先级和运行策略。显然要有更好的性能，在线程数比较多并且线程承担的任务不同的情况下，App中最好还是按任务的类别来划分线程池。

      上面的分析又引出一个问题，我们究竟应该如何配置自己的线程池。

 

       

合理的配置线程池

要想合理的配置线程池，就必须首先分析任务特性，可以从以下几个角度来进行分析：

1. 任务的性质：CPU密集型任务，IO密集型任务和混合型任务。
2. 任务的优先级：高，中和低。
3. 任务的执行时间：长，中和短。
4. 任务的依赖性：是否依赖其他系统资源，如数据库连接。

        任务性质不同的任务可以用不同规模的线程池分开处理。

       CPU密集型任务配置尽可能小的线程，如配置Ncpu+1个线程的线程池。

       IO密集型任务则由于线程并不是一直在执行任务，则配置尽可能多的线程，如2*Ncpu。

       混合型的任务，如果可以拆分，则将其拆分成一个CPU密集型任务和一个IO密集型任务，只要这两个任务执行的时间相差不是太大，那么分解后执行的吞吐率要高于串行执行的吞吐率，如果这两个任务执行时间相差太大，则没必要进行分解。

       我们可以通过Runtime.getRuntime().availableProcessors()方法获得当前设备的CPU个数。

       优先级不同的任务可以使用优先级队列PriorityBlockingQueue来处理。

       它可以让优先级高的任务先得到执行，需要注意的是如果一直有优先级高的任务提交到队列里，那么优先级低的任务可能永远不能执行。

       执行时间不同的任务可以交给不同规模的线程池来处理，或者也可以使用优先级队列，让执行时间短的任务先执行。

       依赖数据库连接池的任务，因为线程提交SQL后需要等待数据库返回结果，如果等待的时间越长CPU空闲时间就越长，那么线程数应该设置越大，这样才能更好的利用CPU。

       建议使用有界队列，有界队列能增加系统的稳定性和预警能力，可以根据需要设大一点，比如几千。

接下来，让我们看看UIL中线程池的配置。

 

 

taskDistributor由于在每创建一个新的线程的时候都需要读取一下磁盘，属于IO操作。需要图片缓存的应用一般在需要加载图片的时候，同时创建很多(>5)线程，这些线程一般来得猛去的也快，存活时间不必太长。

taskExecutor和taskExecutorForCachedImages涉及网络和磁盘的读取和写入操作，比较耗时。主线程数默认为3，实际上IO密集的操作应该定得高一点，以便合理利用CPU的。线程优先级(10为最高，1为最低)为4是比较合理的，因为这些操作只需要后台完成即可，优先级太高可能让界面失去响应。