RxAndroid从零开始学之三（线程调度）

Frank909

发布于 2019-01-14 17:03:47

8340

发布于 2019-01-14 17:03:47

文章被收录于专栏：Frank909Frank909

通过前面的文章，我们已经知道了RxAndroid中的Observable、Subscribers、Observer的创建，及subscribe的使用。接下来，我们开始学习RxAndroid中的线程。

线程

学Android中恐怕无人不知ANR.正因为如此，我们才会使用Thread+Handler或者Aysnctask的代码编写方式。

假设有这么一个需求，程序要进行一个耗时的计算任务，然后得到一个字符串，再显示在界面上的TextView上。用传统方式如下：

new Thread(new Runnable() {
    @Override
    public void run() {
       try {
          Thread.sleep(9000);
        } catch (InterruptedException e) {
          e.printStackTrace();
        }

        runOnUiThread(new Runnable() {
           @Override
           public void run() {
             mTvMsg.setText("找到一个好朋友");
           }
       });
    }
  }).start();

在一个Thread中执行耗时任务，然后通过Handler或者runOnUiThread()方法进行ui操作。代码是比较简单，但太烦琐。后来，Android改进了引入了AyncTask的概念。

class TestTask extends AsyncTask<Void,Void,String>{

    @Override
    protected String doInBackground(Void... params) {
        try {
            Thread.sleep(9000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return "AsyncTask 找到一个好朋友";
    }

    @Override
    protected void onPostExecute(String s) {
        super.onPostExecute(s);
        mTvMsg.setText(s);
    }
 }

@OnClick(R.id.btn_test_aynctask)
public void testAyncTask(){
    TestTask task = new TestTask();
    task.execute();
}

代码简洁了很多，很长一段时间，我们都是以这样的方式编写代码。但现在我们多了一种选择，那就是RxAndroid。我们可以这样写。

Observable.create(new Observable.OnSubscribe<String>() {
    @Override
    public void call(Subscriber<? super String> subscriber) {
        try {
            Thread.sleep(9000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        subscriber.onNext("RxAndroid 找到一个好朋友");
    }
    }).subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())//Android 特有的Thread
    .subscribe(new Action1<String>() {
        @Override
        public void call(String s) {
            mTvMsg.setText(s);
        }
});

说实话，我去年刚接触这个的时候，我在想，这代码也不见得少啊，整天整这些莫名其妙的框架，然后多了很多新概念，有个x用。很多年前，对于mvp模式我也是这样想的。对于一些hello word型的小demo，我干嘛用这么复杂的东西呢？

给自己添堵吗？

但是后来，代码写多了，代码越来越复杂了，如果所有的东西都要自己撸一遍亲自实现，人会整崩溃的，我经常抱着学习的态度去对待项目工程，但有时候边学边写的进度是慢了一点，所以嘛，找个开源的比较可靠的框架应用到项目工程中，可以让自己开发速度加快。当初java语言创立者就说要让程序员写代码像喝咖啡一样轻松，所以才让java语言以咖啡产地爪哇岛为名。而RxAndroid也是这样的一个异步框架，让开发者不再疲于应付多线程异步之间的处理关系。让更多精力。

RxAndroid中线程的处理不在于代码少了多少，而是在于代码结构清楚了不少。这个是很重要的。这个跟mvp模式还真像。

我们看到，从Observable、Subscriber的创建，再到两者之间的subscribe，这就是一条链式的调用，整个事件流程比较清晰，况且Observable的创建，和Subscriber中被观察的地方都可以指定线程，显而易见，广大Android开发者都会喜欢这样的事情，如果可以谁愿意，不厌其烦去写Thread+Handler或者AsyncTask呢？

Schedulers

我们在上一节代码中已经见到过Schedulers.io()这样的形式，它用来调度线程的切换。它的代码不长，方法上都有比较详细的注释。我们可以看看，代码如下：

//代码有删简

/**
 * Static factory methods for creating Schedulers.
 */
public final class Schedulers {

    private final Scheduler computationScheduler;
    private final Scheduler ioScheduler;
    private final Scheduler newThreadScheduler;

    private static final Schedulers INSTANCE = new Schedulers();

    private Schedulers() {

        Scheduler c = RxJavaPlugins.getInstance().getSchedulersHook().getComputationScheduler();
        if (c != null) {
            computationScheduler = c;
        } else {
            computationScheduler = new EventLoopsScheduler();
        }

        Scheduler io = RxJavaPlugins.getInstance().getSchedulersHook().getIOScheduler();
        if (io != null) {
            ioScheduler = io;
        } else {
            ioScheduler = new CachedThreadScheduler();
        }

        Scheduler nt = RxJavaPlugins.getInstance().getSchedulersHook().getNewThreadScheduler();
        if (nt != null) {
            newThreadScheduler = nt;
        } else {
            newThreadScheduler = NewThreadScheduler.instance();
        }
    }

    public static Scheduler immediate() {
        return ImmediateScheduler.instance();
    }


    public static Scheduler newThread() {
        return INSTANCE.newThreadScheduler;
}

    public static Scheduler computation() {
        return INSTANCE.computationScheduler;
    }

    public static Scheduler io() {
        return INSTANCE.ioScheduler;
    }

代码的注释说明Schedulers是一个静态工厂方法类，用来产生各种类型的Scheduler。而且它是单例模式。 Schedulers内部有三个私有的Scheduler变量computationScheduler、ioScheduler、newThreadScheduler。它们分别通过computation()、io()和newThread()暴露。

Schedulers.computation()

/**
     * Creates and returns a {@link Scheduler} intended for computational work.
     * <p>
     * This can be used for event-loops, processing callbacks and other computational work.
     * <p>
     * Do not perform IO-bound work on this scheduler. Use {@link #io()} instead.
     * <p>
     * Unhandled errors will be delivered to the scheduler Thread's {@link java.lang.Thread.UncaughtExceptionHandler}.
     *
     * @return a {@link Scheduler} meant for computation-bound work
     */
    public static Scheduler computation() {
        return INSTANCE.computationScheduler;
    }

代码注释说这是一个计算时用的Scheduler，这里的计算跟cpu相关，主要用于图形计算或者是数据计算，代码特别提到不要在这个线程做与io相关的工作，要用io()代替。

Scheduler.io()

/**
     * Creates and returns a {@link Scheduler} intended for IO-bound work.
     * <p>
     * The implementation is backed by an {@link Executor} thread-pool that will grow as needed.
     * <p>
     * This can be used for asynchronously performing blocking IO.
     * <p>
     * Do not perform computational work on this scheduler. Use {@link #computation()} instead.
     * <p>
     * Unhandled errors will be delivered to the scheduler Thread's {@link java.lang.Thread.UncaughtExceptionHandler}.
     *
     * @return a {@link Scheduler} meant for IO-bound work
     */
    public static Scheduler io() {
        return INSTANCE.ioScheduler;
    }

注释中说明了此方法返回的Scheduler与io操作有关，它的实现是后台的一个按需自动增长的线程池，它常用于异步的阻塞的IO操作，不要在此线程中。

Schedulers.newThread()

/**
     * Creates and returns a {@link Scheduler} that creates a new {@link Thread} for each unit of work.
     * <p>
     * Unhandled errors will be delivered to the scheduler Thread's {@link java.lang.Thread.UncaughtExceptionHandler}.
     *
     * @return a {@link NewThreadScheduler} instance
     */
    public static Scheduler newThread() {
        return INSTANCE.newThreadScheduler;
    }

每次创建一个新的线程去执行代码。

Schedulers.immediate()

/**
     * Creates and returns a {@link Scheduler} that executes work immediately on the current thread.
     * 
     * @return an {@link ImmediateScheduler} instance
     */
    public static Scheduler immediate() {
        return ImmediateScheduler.instance();
    }

这个需要注意的一点是immediate()会让代码立即在当前线程运行。

AndroidSchedulers.mainThread()

在先前的代码中，我也也见过AndroidSchedulers.mainThread()。Android开发中只有UI线程也就是主线程能够操作视图的更新，而RxAndroid中当然就提供这种线程的切换。并且这个Scheduler是为Android定制的。代码非常简单

/** Android-specific Schedulers. */
public final class AndroidSchedulers {
    private AndroidSchedulers() {
        throw new AssertionError("No instances");
    }

    // See https://github.com/ReactiveX/RxAndroid/issues/238
    // https://en.wikipedia.org/wiki/Initialization-on-demand_holder_idiom
    private static class MainThreadSchedulerHolder {
        static final Scheduler MAIN_THREAD_SCHEDULER =
                new HandlerScheduler(new Handler(Looper.getMainLooper()));
    }

    /** A {@link Scheduler} which executes actions on the Android UI thread. */
    public static Scheduler mainThread() {
        Scheduler scheduler =
                RxAndroidPlugins.getInstance().getSchedulersHook().getMainThreadScheduler();
        return scheduler != null ? scheduler : MainThreadSchedulerHolder.MAIN_THREAD_SCHEDULER;
    }
}

内部类保持Looper.getMainLooper（）的Handler的引用，自然就能通过这个Handler去操作主线程。

subscribeOn()

指定向subscribe()动作发生的线程。比如上面例子中的这段代码

public void call(Subscriber<? super String> subscriber) {
    try {
         Thread.sleep(9000);
     } catch (InterruptedException e) {
         e.printStackTrace();
     }
     subscriber.onNext("RxAndroid 找到一个好朋友");
 }

observeOn（）

指定观察者得到通知回调时的线程。如

public void call(String s) {
   mTvMsg.setText(s);
  }

正因为.observeOn(AndroidSchedulers.mainThread())所以mTvMsg.setText(s)才能正常显示。

好了，这篇文章到此为此。简单总结一下。

* RxAndroid提供Scheduler对线程进行切换，不然整个事件会在当前现有的线程上发生，如果是在UI线程直接操作耗时工作或者在非UI线程进行UI视图更新，程序将不会按预期进行*

Schedulers提供了几个方法来获取各类Scheduler

Schedulers.immediate()//当前线程立马执行

Schedulers.newThread()//每次创建新线程运行代码

Scheduler.io()//执行IO操作，或者是网络访问耗时但不耗费CPU的操作

Schedulers.computation() //执行图形计算等复杂计算

Observable的API

subscribeOn()//指定subscribe()发生的线程
observeOn() //观察者得到subscrbe的回调通知时所处的线程

未完待续。。。。

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

原始发表：2016年09月06日，如有侵权请联系 cloudcommunity@tencent.com 删除

android

java

本文分享自作者个人站点/博客前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体分享计划，欢迎热爱写作的你一起参与！

android

java

登录后参与评论

0 条评论

热度