AsyncTask中的SerialExecutor

ゞ浴缸里的玫瑰 2023-07-03 02:35 25阅读 0赞

## AsyncTask的线程模型 ##

AsyncTask专门用来进行UI线程与工作线程之间通信。每个AsyncTask只能执行一个任务，执行多个任务时需要创建多个AsyncTask实例。最早版本的AsyncTask，多个实例共享一个后台线程，所以多个task只能串行。

从1.6开始，AsyncTask变成了共享线程池，彼时的线程池定义如下：

//核心线程数
    private static final int CORE_POOL_SIZE = 5;
    //最大线程数
    private static final int MAXIMUM_POOL_SIZE = 128;
    //等待时间
    private static final int KEEP_ALIVE = 1;
    //阻塞队列
    private static final BlockingQueue<Runnable> sWorkQueue =
            new LinkedBlockingQueue<Runnable>(10);
    //线程池
    private static final ThreadPoolExecutor sExecutor = new ThreadPoolExecutor(CORE_POOL_SIZE,
            MAXIMUM_POOL_SIZE, KEEP_ALIVE, TimeUnit.SECONDS, sWorkQueue, sThreadFactory);

虽然线程池可以实现多个任务的并行执行，但最大线程数128，如果有超过128个任务同时提交则会造成崩溃。Google意识到AsyncTask滥用带来的问题，3.0之后引入SerialExecutor作为线程池的默认实现，SerialExeutor让AsyncTask的并行执行又变为了串行，同一时间只有一个Task在后台运行。

也许有人会问那直接将线程池改回线程不就好了吗？这是为了保证接口的灵活性，使用者如果希望task并行执行时，可以随时用自定义线程池替换默认的SerialExecutor

## SerialExecutor的实现 ##

SerialExecutor是如何保证串行的呢？

public static final Executor THREAD_POOL_EXECUTOR;
    
    private static final BlockingQueue<Runnable> sPoolWorkQueue =
                new LinkedBlockingQueue<Runnable>(128);
    
    static {
            ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(
                    CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, KEEP_ALIVE_SECONDS, TimeUnit.SECONDS,
                    sPoolWorkQueue, sThreadFactory);
            threadPoolExecutor.allowCoreThreadTimeOut(true);
            THREAD_POOL_EXECUTOR = threadPoolExecutor;
    }
    
    private static class SerialExecutor implements Executor {
            final ArrayDeque<Runnable> mTasks = new ArrayDeque<Runnable>();
            Runnable mActive;
    
            public synchronized void execute(final Runnable r) {
                mTasks.offer(new Runnable() {
                    public void run() {
                        try {
                            r.run();
                        } finally {
                            scheduleNext();
                        }
                    }
                });
                if (mActive == null) {
                    scheduleNext();
                }
            }
    
            protected synchronized void scheduleNext() {
                if ((mActive = mTasks.poll()) != null) {
                    THREAD_POOL_EXECUTOR.execute(mActive);
                }
            }
        }

内部维护了一个 无限大小的队列mTasks，AsyncTask的任务不直接调用线程池执行（THREAD\_POOL\_EXECUTOR.execute） ，而是添加到 mTasks队列中。然后按照FIFO的原则通过scheduleNext还行第一个任务。任务执行完后通过finally\{...\}驱动scheduleNext执行下一个，直到队列任务全部清空。

## SerialExecutor的意义 ##

随着Coroutine、RxJava等优秀框架的出现，AsyncTask的使用率越来越低了，但是SerialExecutor的使用需求却一直存在：

例如有5个任务，Runnable1 ...  5，其中Runnable1、3、5和Runnable2、4分别希望以串行顺序执行

//单线程线程池：可以并行的地方也要串行执行，效率低
    executor.execute(runnable1)
    executor.execute(runnable3)
    executor.execute(runnable5)
    executor.execute(runnable2)
    executor.execute(runnable4)
    
    //多线程线程池：需要新额外创建Runnalbe用来包裹串行任务，不优雅
    executor.execute {
        runnable1.run()
        runnable3.run()
        runnable5.run()
    }
    executor.execute {
        runnable2.run()
        runnable4.run()
    }

上述两种方法都可以实现需求，但是都不完美，如果我们使用SerialExecutor

executor1.execute(runnable1)
    executor1.execute(runnable3)
    executor1.execute(runnable5)
    executor2.execute(runnable2)
    executor2.execute(runnable4)

executor1和executor2都是SerialExecutor实例，每个executor内部保证任务的串行执行，多个executor又共享线程池，实现资源最大有效利用。

## SerialExecutor使用的注意及改进 ##

AsyncTask中所有Task共享一个SerialExecutor实例，所有任务都是串行。如果脱离AsyncTask使用SerialExecutor则有可能同时有多个实例运行。此时需要注意SerialExecutor共享的线程池THREAD\_POOL\_EXECUTOR，线程池的队列mPoolWrokQueue为128，也就是说超过128个SerialExecutor存在时有可能会造成Crash。一个简单的办法就是将mPoolWrokQueue改造为无界队列

private val mPoolWrokQueue = LinkedBlockingQueue<Runnable>()

但是因为队列是无穷大的，所以MAXIMUM\_POOL\_SIZE可能永远无法达到，线程池无法达到理论最大吞吐量。

我们将队列改造如下：

private val THREAD_POOL_EXECUTOR by lazy {
            ThreadPoolExecutor(
                    CORE_POOL_SIZE, MAXIMUM_POOL_SIZE, 0L, TimeUnit.SECONDS, ThreadPoolQueue(this)
                )
    }
    
    class ThreadPoolQueue(val executor: ThreadPoolExecutor) : LinkedBlockingQueue<Runnable>() {
          override fun offer(e: Runnable?): Boolean {
                    return executor.run {
                        if (activeCount in poolSize until maximumPoolSize) false
                        else super.offer(e)
                    }
           }
    }

向队列添加任务时会根据当前线程数动态的返回结果：如果可以去启动新线程则优先启动新线程，达到最大线程数时再真正向队列添加。