RxJava 学习（一）

向右看齐 2024-02-17 18:17 50阅读 0赞

在Android Studio中使用RxJava

先添加依赖

compile 'io.reactivex:rxjava:1.2.1'
        compile 'io.reactivex:rxandroid:1.2.1'

之后，就可以正常使用RxJava了

RxJava到底是什么，干什么用的呢

RxJava是一个已观察者模式来写的一个框架，主要应用就是异步处理程序上的问题。

为什么要异步呢，当然就是保证主线程不被卡住，把费时的逻辑都放在另外一个线程中来处理。

那为什么不用Thread，ThreadPool，AsyncTask呢

可以用啊，不过RxJava更省事，代码更清晰

那么观察者模式是什么呢，观察者模式是又一个观察者，一个被观察者，观察者订阅被观察者的一些事件，

当对应事件触发的时候，观察者会得到被观察者的通知。

RxJava也是这个模式

先来定义一个观察者Observer

Observer<String> observer = new Observer<String>() {
            @Override
            public void onCompleted() {
    
            }
    
            @Override
            public void onError(Throwable e) {
    
            }
    
            @Override
            public void onNext(String s) {
    
            }
        };

也可以定一个Subscriber，它是实现Observer接口的一个抽象类

Subscriber<String> subscriber = new Subscriber<String>() {
            @Override
            public void onCompleted() {
    
                Log.e("onCompleted", "completed");
    
            }
    
            @Override
            public void onError(Throwable e) {
    
                e.printStackTrace();
    
            }
    
            @Override
            public void onNext(String s) {
    
                Log.e("onNext", s);
    
            }
        };

Subscriber比Observer多了两个方法可以使用，一个是onStart，在开始订阅的时候触发，

还有一个unsubscribe，用于取消订阅

其他用法一样，一般，我们都会定义一个Subscriber来使用

再来定义一个被观察者

Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
            @Override
            public void call(Subscriber<? super String> subscriber) {
                subscriber.onNext("a");
                subscriber.onNext("b");
                subscriber.onCompleted();
    
            }
        });

当被观察者被订阅的时候，会自动触发call方法，在call中我们处理自己的逻辑，可以随时出发观察者的

onNext方法，当onCompleted方法被触发的时候，表示本次订阅结束。

那怎么产生订阅关系呢，

observable.subscribe(subscriber);

这就产生了订阅的关系。

还有，如果被观察者只是我上面例子中的简单形式，也可以这样定义

Observable<String> observable1 = Observable.just("a", "b");

或者这样

String[] str = {"a", "b"};
        Observable<String> observable2 = Observable.from(str);

观察者也可以定义成不完全的，只实现某个方法

Action1<String> onNext = new Action1<String>() {
            @Override
            public void call(String s) {
    
            }
        };
        Action1<Throwable> onError = new Action1<Throwable>() {
            @Override
            public void call(Throwable throwable) {
    
            }
        };
        Action0 onComplete = new Action0() {
            @Override
            public void call() {
    
            }
        };

像这样

订阅就改成这样了

observable.subscribe(onNext);
                    observable.subscribe(onNext, onError);
                    observable.subscribe(onNext, onError, onComplete);

观察者，被观察者，订阅的基本写法也就这些

然并卵

因为所有事情都是在同一个线程里进行的，完全没有异步，这么写也就没什么意义了

那怎么来定义异步呢

在RxJava里，通过Scheduler可以实现线程的调度

//当前线程
                    Schedulers.immediate();
                    //新启动线程
                    Schedulers.newThread();
                    //IO专用线程
                    Schedulers.io();
                    //计算专用线程
                    Schedulers.computation();
                    //Android主线程
                    AndroidSchedulers.mainThread();

通过，这样来定义线程

observable.observeOn(AndroidSchedulers.mainThread()) //定义观察者的线程
                            .subscribeOn(Schedulers.io()) //定义被观察者的线程
                            .subscribe(subscriber);

RxJava的基本用法也就是这样了