RxJava 线程的调度

1.Android为例, 一个Activity的所有动作默认都是在主线程中运行的, 所有耗时操作要在线程实现. 所以在开发的时候,经常来来切换线程. 导致项目的代码的难以维护. 而Rxjava 很好的解决这个线程切换.
在RxJava 在切换线程时用到了两个方法 subscribeOn() 和 observeOn() 下面来分别解释一下这两个方法

subscribeOn:

影响的是最开始的被观察者所在的线程。当使用多个 subscribeOn() 的时候，只有第一个 subscribeOn() 起作用.
observeOn:
影响的是跟在后面的操作（指定观察者运行的线程）。所以如果想要多次改变线程，可以多次使用 ####observeOn；

默认情况下，事件序列操作的线程与调用.subscribe()的线程一致

写一个栗子来实践一下.

    //创建一个被观察者
        Observable<Integer> integerObservable = Observable.create(new ObservableOnSubscribe<Integer>() {
            @Override
            public void subscribe(ObservableEmitter<Integer> emitter) throws Exception {
                Log.d("---->", "我在什么线程" + Thread.currentThread().getName());
                Log.d("---->", "发射数据" + 1);
                emitter.onNext(1);

            }
        });

            //创建观察者

        Observer<Integer>   observer =      new    Observer<Integer>(){
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer integer) {
                Log.d("---->", "我在什么线程" + Thread.currentThread().getName());

                Log.d("---->","结果"+integer);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        };

        //创建联系
        integerObservable.subscribe(observer);

在主线程中分别创建被观察者和观察者, 然后将他们连接在一起, 同时分别打印出它们所在的线程, 运行结果为:

06-01 20:13:40.100 1441-1441/view.dome.com.rxjavadome D/---->: 我在什么线程main
    发射数据1
    我在什么线程main
    结果1

说明默认是在同一个线程工作.
这样肯定是满足不了我们的需求的, 我们更多想要的是这么一种情况, 在子线程中做耗时的操作, 然后回到主线程中来操作UI.

当使用了 RXJava 里面

  //创建联系
        integerObservable
                .subscribeOn(Schedulers.newThread())
                .observeOn(AndroidSchedulers.mainThread())
                .subscribe(observer);

结果就不一样了

06-01 20:25:00.210 1657-1671/view.dome.com.rxjavadome D/---->: 我在什么线程RxNewThreadScheduler-1
    发射数据1
06-01 20:25:00.330 1657-1657/view.dome.com.rxjavadome D/---->: 我在什么线程main
    结果1

发现第一次进那个是新的线程,而结果是的数据在还在主线,
当我在subscribeOn 里面多次切换线程, observeOn 在里面也多次切换线程会是怎么样.

   //创建联系
        integerObservable
                .subscribeOn(Schedulers.newThread())
                .subscribeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())
                .observeOn(Schedulers.io())
                .subscribe(observer);

结果:

06-01 20:38:23.930 1756-1771/view.dome.com.rxjavadome D/---->: 我在什么线程RxNewThreadScheduler-1
    发射数据1
06-01 20:38:24.070 1756-1772/view.dome.com.rxjavadome D/---->: 我在什么线程RxCachedThreadScheduler-2
    结果1

发现只是最后的一个线程起作用了,

当我们把observeOn 里面的线程位置换一下,看一下

  integerObservable
                .subscribeOn(Schedulers.newThread())
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.io())
                .observeOn(AndroidSchedulers.mainThread())

                .subscribe(observer);

结果

06-01 20:41:42.510 1823-1838/view.dome.com.rxjavadome D/---->: 我在什么线程RxNewThreadScheduler-1
    发射数据1
06-01 20:41:42.630 1823-1823/view.dome.com.rxjavadome D/---->: 我在什么线程main
    结果1

Schedulers.io() 代表io操作的线程, 通常用于网络,读写文件等io密集型的操作
Schedulers.computation() 代表CPU计算密集型的操作, 例如需要大量计算的操作
Schedulers.newThread() 代表一个常规的新线程
AndroidSchedulers.mainThread() 代表Android的主线程

在Retrofit2.0 跟RxJava 结合使用.

gradle 的依赖添加

       // Android 支持 Rxjava
    implementation 'io.reactivex.rxjava2:rxjava:2.1.14'
    implementation 'io.reactivex.rxjava2:rxandroid:2.0.2'
    // Retrofit库
    implementation 'com.squareup.retrofit2:retrofit:2.4.0'
    // Okhttp库
    implementation 'com.squareup.okhttp3:okhttp:3.10.0'
    //Gson解析
    implementation'com.squareup.retrofit2:converter-gson:2.4.0'
    // 此处一定要注意使用RxJava2的版本
    implementation 'com.jakewharton.retrofit:retrofit2-rxjava2-adapter:1.0.0'
    //Log
    implementation 'com.squareup.okhttp3:logging-interceptor:3.9.0'

2.数据结构.

{
    "status":1,
    "content":{
        "from":"en-EU",
        "to":"zh-CN",
        "vendor":"tencent",
        "out":"嗨世界",
        "ciba_use":"来自机器翻译。",
        "ciba_out":"",
        "err_no":0
    }
}

编写数据Bean

package view.dome.com.rxjavadome.bean;


/**
 * 数据bean
 */
public class Translation {
    
    /**
     * status : 1
     * content : {"from":"en-EU","to":"zh-CN","vendor":"tencent","out":"嗨世界","ciba_use":"来自机器翻译。","ciba_out":"","err_no":0}
     */

    private int status;
    private ContentBean content;
    
    public int getStatus() {
        return status;
    }

    public void setStatus(int status) {
        this.status = status;
    }

    public ContentBean getContent() {
        return content;
    }

    public void setContent(ContentBean content) {
        this.content = content;
    }


    public static class ContentBean {
        /**
         * from : en-EU
         * to : zh-CN
         * vendor : tencent
         * out : 嗨世界
         * ciba_use : 来自机器翻译。
         * ciba_out :
         * err_no : 0
         */

        private String from;
        private String to;
        private String vendor;
        private String out;
        private String ciba_use;
        private String ciba_out;
        private int err_no;

        public String getFrom() {
            return from;
        }

        public void setFrom(String from) {
            this.from = from;
        }

        public String getTo() {
            return to;
        }

        public void setTo(String to) {
            this.to = to;
        }

        public String getVendor() {
            return vendor;
        }

        public void setVendor(String vendor) {
            this.vendor = vendor;
        }

        public String getOut() {
            return out;
        }

        public void setOut(String out) {
            this.out = out;
        }

        public String getCiba_use() {
            return ciba_use;
        }

        public void setCiba_use(String ciba_use) {
            this.ciba_use = ciba_use;
        }

        public String getCiba_out() {
            return ciba_out;
        }

        public void setCiba_out(String ciba_out) {
            this.ciba_out = ciba_out;
        }

        public int getErr_no() {
            return err_no;
        }

        public void setErr_no(int err_no) {
            this.err_no = err_no;
        }
    }
}

3.写Observable<..>接口形式

public interface GetRequest_Interface {
    @GET("ajax.php?a=fy&f=auto&t=auto&w=hi%20world")
    Observable<Translation> getCall();
    // 注解里传入 网络请求 的部分URL地址
    // Retrofit把网络请求的URL分成了两部分：一部分放在Retrofit对象里，另一部分放在网络请求接口里
    // 如果接口里的url是一个完整的网址，那么放在Retrofit对象里的URL可以忽略
    // 采用Observable<...>接口
    // getCall()是接受网络请求数据的方法
}

4.编写Retrofit跟Rx结合使用

 //定制OkHttp
        OkHttpClient.Builder httpClientBuilder = new OkHttpClient
                .Builder();
        if (BuildConfig.DEBUG) {//发布版本不再打印
            // 日志显示级别
            HttpLoggingInterceptor.Level level= HttpLoggingInterceptor.Level.BODY;
            //新建log拦截器
            HttpLoggingInterceptor loggingInterceptor=new HttpLoggingInterceptor(new HttpLoggingInterceptor.Logger() {
                @Override
                public void log(String message) {
                Log.e("Tag",message);
                }
            });
            loggingInterceptor.setLevel(level);
            //OkHttp进行添加拦截器loggingInterceptor
            httpClientBuilder.addInterceptor(loggingInterceptor);

        }
        OkHttpClient client = httpClientBuilder.build();
        Retrofit retrofit = new Retrofit.Builder() //创建retrofit
                .baseUrl("http://fy.iciba.com/") // 设置 网络请求 Url
                .addConverterFactory(GsonConverterFactory.create()) //设置使用Gson解析(记得加入依赖)
                .addCallAdapterFactory(RxJava2CallAdapterFactory.create()) // 支持RxJava
                .client(client) //添加拦截器
                .build();

        //  创建 网络请求接口 的实例
        GetRequest_Interface request = retrofit.create(GetRequest_Interface.class);
        //采用Observable<...>形式 对 网络请求 进行封装
        Observable<Translation> observable = request.getCall();

        observable.subscribeOn(Schedulers.io()) //在子线程请求网络
                .observeOn(AndroidSchedulers.mainThread()) //在主线更新数据
                .subscribe(new Observer<Translation>() {
                    @Override
                    public void onSubscribe(Disposable d) {
                        Log.d("----->", "开始采用subscribe连接");
                    }

                    @Override
                    public void onNext(Translation translation) {
                        Log.d("----->", "对返回的数据进行处理");

                        if (translation.getStatus()==1){   //成功

                            Translation.ContentBean content = translation.getContent();
                            if (content!=null){
                                String ciba_out = content.getOut();
                                Toast.makeText(MainActivity.this,ciba_out,Toast.LENGTH_SHORT).show();
                            }
                        }else {
                            Toast.makeText(MainActivity.this,"请求失败",Toast.LENGTH_SHORT).show();

                        }

                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.d("----->", "请求失败"+e.toString());

                    }

                    @Override
                    public void onComplete() {
                        Log.d("----->", "请求成功");
                    }
                });

RxJava 线程的调度

subscribeOn:

推荐阅读更多精彩内容