Android上的线程池使用

Android线程 特性

1. Each thread costs a minimum of 64k of memory.
2. Generally, threads in the foreground group get about 95% of the total execution time from the device, while the background group gets roughly 5%.
3. 主线程的stackSize 是8M
4. 普通线程的stackSize 是1M左右 （根据ANR可以看到主要是 1038/1014 KB）

基本线程规划

Io线程

负责执行耗时的io操作

例如：网络请求，文件读写，资源加载等

线程数量: CPU_COUNT

Computation线程

负责计算相关操作

例如：json解析，大集合对象遍历，bitmap处理等

线程数量: CPU_COUNT

好处：

1. 语义上非常直白

2. Io密集型操作和计算密集型操作是多线程调度中的重要区别，分别对待可以更好的针对不同的任务属性进行优化。IO线程池数量更多，但是优先级更低一些，Computation 更少，但是优先级更高

3. Io 和 Computation分开还可以防止出现Io任务阻塞住线程，Computation线程得不到执行的情况

基于RxJava的线程调度API

1. 使用RxJava 的 observeOn 和 subscribeOn 进行线程切换

2. 使用RxJava 的Scheduelr抽象作为并行执行器

具体来说是：
Schedulers.io 和 Schedulers.computation

最佳实践

1. 凡事涉及到IO操作的，例如文件，网络，都分派到io执行器；
2. 耗时执行的非io操作，都分派到computation线程
3. 熟悉RxJavaScheduler实现和线程切换特性（下面单独介绍）
4. 针对现有的基于Executor的代码，使用RxExecutors进行修改（单独介绍）,Executor接口更方便的场景也可以使用RxExecutors
5. 谨慎使用Schedulers.newThread() ，最好不要使用，现在使用了RxJava的hook机制检查了这种情况的调用，会有warn级别的log

RxExecutors

1. 实现了Executor接口

2. 基于RxJava的Scheduler实现

3. 存在两个实例，Io， Computation 同RxJava保持一致

    public enum RxExecutors implements Executor {
   
    Io {
        @Override
        Scheduler.Worker createWorker() {
            return Schedulers.io().createWorker();
        }
    },
   
    Computation {
        @Override
        Scheduler.Worker createWorker() {
            return Schedulers.computation().createWorker();
        }
    };
   

RxExecutors API

• submit(Runnable) -> Subscription
• submit(Action0) -> Subscription
• delay(Action0,int delay, TimeUnit) -> Subscription
• delay(Runnable, int delay, TimeUnit) -> Subscription
• schedulePeriodically(Action0,int initialDelay, int peroid, TimeUnit) -> Subscription
• schedulePeriodically(Runnable,int initialDelay, int peroid, TimeUnit) -> Subscription
• execute(Runnable) -> void

简单直白的API

使用示例

//在线程池中执行任务
RxExecutors.Computation.execute(runnable);
RxExecutors.Io.submit(runnable);
// 执行延时任务
RxExecutors.Computation.delay(runnable, 1, TimeUnit.SECONDS);
// 执行周期性任务
RxExecutors.Io.schedulePeriodically(runnable, 1, 1, TimeUnit.SECONDS);

针对Android平台的优化

这两个优化和bolts-task一致

RxJava IO 线程池优化

RxJava IO 线程池是cache类型的线程池，短时间大量任务的情况下，会创建非常多的线程，对Android平台不够友好

public static final Supplier<ExecutorService> IO_THREAD_POOL = Suppliers.synchronizedSupplier(Suppliers.memoize(new Supplier() {
        public ExecutorService get() {
            return AndroidExecutors.newCacheThreadPool(new CustomThreadFactory("RxIoScheduler-", 10, false));
        }
    }));

 public static void optimizeSchedulerForAndroid() {
        RxJavaPlugins.getInstance().registerSchedulersHook(new RxJavaSchedulersHook() {
            public Scheduler getIOScheduler() {
                return Schedulers.from((Executor)RxCustomize.IO_THREAD_POOL.get());
            }
        });
    }

不改变RxJava API的情况下 针对Android 平台做了最大线程数目限制的线程池 线程可以超期释放

核心线程可释放

默认情况下，线程池核心线程是常驻的

    public static ExecutorService newCacheThreadPool(ThreadFactory threadFactory) {
        ThreadPoolExecutor executor = new ThreadPoolExecutor(CPU_COUNT, MAX_CPU_COUNT, 1L, TimeUnit.SECONDS, new LinkedBlockingQueue(), threadFactory);
        executor.allowCoreThreadTimeOut(true);
        return executor;
    }

RxJava

RxJava的Scheduler特性

Schedulers.newThread()

• 会开启新的线程执行任务

Schedulers.io()

• 类似cache类型的‘线程池’,数量没有上限，不太适合移动设备；

• 我们已经使用RxJava的Plugin机制，进行了优化，使用固定数量（CPU_COUNT）的线程池替代，对上层API没有影响，感兴趣的同学可以看一下RxCustomize的实现

Schedulers.computation()

• 类似固定数量（CPU_COUNT）的线程池

Schedulers.immediate()

• 在当前线程执行，delay的操作会使用Thread.sleep()

Schedulers.trampoline()

• 同样在当前线程进行执行，但是会加入按照执行时间和加入顺序进行排序的优先级队列，从队列头开始执行任务

RxJava 的线程切换API

参考Scheduler

If you want to introduce multithreading into your cascade of Observable operators, you can do so by instructing those operators (or particular Observables) to operate on particular Schedulers.

Some ReactiveX Observable operators have variants that take a Scheduler as a parameter. These instruct the operator to do some or all of its work on a particular Scheduler.

By default, an Observable and the chain of operators that you apply to it will do its work, and will notify its observers, on the same thread on which its Subscribe method is called. The SubscribeOn operator changes this behavior by specifying a different Scheduler on which the Observable should operate. The ObserveOn operator specifies a different Scheduler that the Observable will use to send notifications to its observers.

As shown in this illustration, the SubscribeOn operator designates which thread the Observable will begin operating on, no matter at what point in the chain of operators that operator is called. ObserveOn, on the other hand, affects the thread that the Observable will use below where that operator appears. For this reason, you may call ObserveOn multiple times at various points during the chain of Observable operators in order to change on which threads certain of those operators operate.

总结来说：

1. 默认情况下，整条Obserber链条运行在执行subscribe所在的线程（变现线程）
2. subscribeOn可以改变Obserber线程,可以整条操作链条的线程，并且不管在整条操作链条任何点调用都在整条链条起作用，多次调用，只有第一次起作用
3. observeOn 可以影响接下来的操作的执行线程，每次调用都起作用，都影响接下的调用
4. subscribeOn比oberseOn影响范围更大
5. 掌握好RxJava的线程切换语义非常关键