RxJava基本使用

2017-08-13  本文已影响66人  自然like
image
参考：给 Android 开发者的 RxJava 详解
GitHub：RxJava、RxAndroid 
// Build.gradle配置文件引入
compile 'io.reactivex:rxandroid:1.1.0'
compile 'io.reactivex:rxjava:1.1.0'

1、RxJava可以做什么

RxJava是一个在 Java VM 上使用可观测的序列来组成异步的、基于事件的程序的库。可以轻松实现线程切换的方式。平时我们做线程切换的几种方式：

handler.sendEmptyMessage(EVENT_DOWNLOAD_DATA);

Handler另外的一种发送消息的方式：

getUiHandler().post(new Runnable() {
    @Override
    public void run() {
        // do something                
    }
});

getActivity().runOnUiThread(new Runnable() {
    @Override
    public void run() {
         // do something       
    }
});

相必，以上的几种方式使用起来都不是那么的简洁，业务复杂起来，还会导致逻辑上的不清晰，那么RxJava就可以做到逻辑的简洁，执行异步任务、线程切换都可以在一条链上完成。

2、观察者模式

观察者模式的例子：

3、RxJava观察者模式实现步骤

3.1、创建观察者 Observer

Observable被观察者 和 Observer观察者 通过 subscribe() 方法实现订阅关系
事件。

onNext()
onCompleted()
onError()
onStart() // 这是 Subscriber 增加的方法。它总是在 subscribe 所发生的线程被调用。不太适用于执行任务前的弹窗等UI操作。
unsubscribe() // 这是 Subscriber 所实现的另一个接口 Subscription 的方法，用于取消订阅。避免内存泄漏。

Observer 接口，Observer 的抽象类：Subscriber

Subscriber<String> subscriber = new Subscriber<String>() {
    @Override
    public void onNext(String s) {
        Log.d(tag, "Item: " + s);
    }

    @Override
    public void onCompleted() {
        Log.d(tag, "Completed!");
    }

    @Override
    public void onError(Throwable e) {
        Log.d(tag, "Error!");
    }
};
3.2、创建被观察者 Observable

create() 方法是 RxJava 最基本的创造事件序列的方法

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

另外还提供了一些方法用来快捷创建事件队列

// just()
Observable observable = Observable.just("Hello", "Hi", "Aloha");

// from()
String[] words = {"Hello", "Hi", "Aloha"};
Observable observable = Observable.from(words);
3.3、Subscribe (订阅)
observable.subscribe(observer);
// 或者：
observable.subscribe(subscriber);

观察者被观察者换位置：对流式 API 的设计会造成影响。

在 RxJava 中， Observable 并不是在创建的时候就立即开始发送事件，而是在它被订阅的时候。

除了 subscribe(Observer) 和 subscribe(Subscriber) ，subscribe() 还支持不完整定义的回调

// 自动创建 Subscriber ，并使用 onNextAction 来定义 onNext()
observable.subscribe(onNextAction);
// 自动创建 Subscriber ，并使用 onNextAction 和 onErrorAction 来定义 onNext() 和 onError()
observable.subscribe(onNextAction, onErrorAction);
// 自动创建 Subscriber ，并使用 onNextAction、 onErrorAction 和 onCompletedAction 来定义 onNext()、 onError() 和 onCompleted()
observable.subscribe(onNextAction, onErrorAction, onCompletedAction);

Action0 是 RxJava 的一个接口，它只有一个方法 call()，这个方法是无参无返回值的；由于 onCompleted() 方法也是无参无返回值的，因此 Action0 可以被当成一个包装对象，将 onCompleted() 的内容打包起来将自己作为一个参数传入 subscribe() 以实现不完整定义的回调。这样其实也可以看做将 onCompleted() 方法作为参数传进了 subscribe()，相当于其他某些语言中的『闭包』。

4、RxJava操作符

变换：就是将事件序列中的对象或整个序列进行加工处理，转换成不同的事件或事件序列

FuncX 和 ActionX 的区别在 FuncX 包装的是有返回值的方法。

4.1、map()

事件对象的直接变换

Observable.from(students)
    .map(new Func1<Student, String>() {
        @Override
        public String call(Student student) {
            return student.getName();
        }
    })
    .subscribe(subscriber);
4.2、flatMap()
Observable.from(students)
    .flatMap(new Func1<Student, Observable<Course>>() {
        @Override
        public Observable<Course> call(Student student) {
            return Observable.from(student.getCourses());
        }
    })
    .subscribe(subscriber);

flatMap() 中返回的是个 Observable 对象，并且这个 Observable 对象并不是被直接发送到了 Subscriber 的回调方法中。 flatMap() 的原理是这样的：

由于可以在嵌套的 Observable 中添加异步代码， flatMap() 也常用于嵌套的异步操作，例如嵌套的网络请求。

5、Scheduler 线程切换

Observable.just(1, 2, 3, 4) // IO 线程，由 subscribeOn() 指定
    .subscribeOn(Schedulers.io())
    .observeOn(Schedulers.newThread())
    .map(mapOperator) // 新线程，由 observeOn() 指定
    .observeOn(Schedulers.io())
    .map(mapOperator2) // IO 线程，由 observeOn() 指定
    .observeOn(AndroidSchedulers.mainThread) 
    .subscribe(subscriber);  // Android 主线程，由 observeOn() 指定

通过 observeOn() 的多次调用，程序实现了线程的多次切换。

不过，不同于 observeOn() ， subscribeOn() 的位置放在哪里都可以，但它是只能调用一次的。

与 Subscriber.onStart() 相对应的，有一个方法 Observable.doOnSubscribe() 。它和 Subscriber.onStart() 同样是在 subscribe() 调用后而且在事件发送前执行，但区别在于它可以指定线程。默认情况下， doOnSubscribe() 执行在 subscribe() 发生的线程；而如果在 doOnSubscribe() 之后有 subscribeOn() 的话，它将执行在离它最近的 subscribeOn() 所指定的线程。

Observable.create(onSubscribe)
    .subscribeOn(Schedulers.io())
    .doOnSubscribe(new Action0() {
        @Override
        public void call() {
            progressBar.setVisibility(View.VISIBLE); // 需要在主线程执行
        }
    })
    .subscribeOn(AndroidSchedulers.mainThread()) // 指定主线程
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(subscriber);
上一篇 下一篇

猜你喜欢

热点阅读