Android框架学习之RxJava(一)基本概念

**

前言

之前学习安卓只知道去看，但是理解之后时间一长就遗忘了，没有养成一个好习惯，之前有做笔记的习惯，但是过于麻烦，以后就用写博客的形式将自己学习的框架知识积累起来，希望有所进步。 **

添加依赖

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

一.基本概念

1.Observer：观察者，源码中Observer是一个接口，其中有 onCompleted()，onError(Throwable e)，onNext(T t)方法。

public interface Observer<T> { /** * Notifies the Observer that the {@link Observable} has finished sending push-based notifications. * <p> * The {@link Observable} will not call this method if it calls {@link #onError}. */ void onCompleted(); /** * Notifies the Observer that the {@link Observable} has experienced an error condition. * <p> * If the {@link Observable} calls this method, it will not thereafter call {@link #onNext} or * {@link #onCompleted}. * * @param e * the exception encountered by the Observable */ void onError(Throwable e); /** * Provides the Observer with a new item to observe. * <p> * The {@link Observable} may call this method 0 or more times. * <p> * The {@code Observable} will not call this method again after it calls either {@link #onCompleted} or * {@link #onError}. * * @param t * the item emitted by the Observable */ void onNext(T t); }

2.Subscription:订阅，Observable调用subscribe( )方法返回的对象，可以用来取消订阅事件，源码中Subscription是一个接口，其中有unsubscribe()，isUnsubscribed()两个方法，rxjava中好像有自动取消订阅的机制。

public interface Subscription { /** * Stops the receipt of notifications on the {@link Subscriber} that was registered when this Subscription * was received. * <p> * This allows unregistering an {@link Subscriber} before it has finished receiving all events (i.e. before * onCompleted is called). */ void unsubscribe(); /** * Indicates whether this {@code Subscription} is currently unsubscribed. * * @return {@code true} if this {@code Subscription} is currently unsubscribed, {@code false} otherwise */ boolean isUnsubscribed(); }

3.Subscriber：订阅者。Subscriber是实现Observer和Subscription接口的一个抽象类。

public abstract class Subscriber<T> implements Observer<T>, Subscription

4.Observable：被观察者，它决定什么时候触发事件以及触发怎样的事件， RxJava 使用 create() 方法来创建一个 Observable ，并为它定义事件触发规则。

5.subscribeOn（）：主要改变的是订阅的线程，即call()执行的线程（该方法调用前序列所运行的线程），只能调用一次，多次调用只以第一个为准，作用于整个调用过程。

6.observeOn（）：主要改变的是发送的线程，即onNext()执行的线程（该方法调用后序列所运行的线程），这里的onNext()是new Subscriber.onNext()，可以多次调用，多次改变不同的接受者所在的scheduler，作用于observeOn调用之后。

Observable.create(new Observable.OnSubscribe<String>() { @Override public void call(Subscriber<? super String> subscriber) { //subscriber.onStart(); 这个方法在订阅后自动执行 subscriber.onNext("Hello RxAndroid"); subscriber.onCompleted(); } }).subscribeOn(Schedulers.newThread()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Subscriber<String>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(String s) { } @Override public void onStart() { } });

7.Scheduler:调度程序,用来调度程序运行的线程的。

(1)Schedulers.immediate(): 在当前线程运行，相当于不指定线程，不设置就默认这个。

(2)Schedulers.newThread(): 总是启用新线程，并在新线程执行操作。

(3)Schedulers.io(): I/O 线程，常用于一些I/O操作，例如读写文件、读写数据库、网络信息交互等，和newThread()差不多，只是io()内部实现用了一个无数量上限的线程池，可以重用空闲的线程，所以比newThread()相率高一点。

(4) Schedulers.computation(): 计算所使用的 Scheduler。

(5) AndroidSchedulers.mainThread()：RxAndroid钟特有的，为安卓的主线程，也就是UI线程。

8.subscribe()：订阅事件，用来订阅Observable。订阅中可以是Action1，Subscriber，Observer或者SingleSubscriber。

//使用Observable.fromCallable()方法有两点好处： //获取要发送的数据的代码只会在有Observer订阅之后执行。 //获取数据的代码可以在子线程中执行。 Observable.fromCallable(new Callable<Object>() { @Override public Object call() throws Exception { return null;//在这里执行方法 } }).subscribeOn(Schedulers.newThread()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Action1<Object>() { @Override public void call(Object o) { } }, new Action1<Throwable>() { @Override public void call(Throwable throwable) { } }, new Action0() { @Override public void call() { } }); //Subscriber是接口Observer 的实现类 Observable.create(new Observable.OnSubscribe<String>() { @Override public void call(Subscriber<? super String> subscriber) { //subscriber.onStart(); 这个方法在订阅后自动执行 subscriber.onNext("Hello RxAndroid"); subscriber.onCompleted(); } }).subscribeOn(Schedulers.newThread()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Subscriber<String>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(String s) { } @Override public void onStart() { } }); //Observer Observable.create(new Observable.OnSubscribe<String>() { @Override public void call(Subscriber<? super String> subscriber) { //subscriber.onStart(); 这个方法在订阅后自动执行 subscriber.onNext("Hello RxAndroid"); subscriber.onCompleted(); } }).subscribeOn(Schedulers.newThread()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Observer<String>() { @Override public void onCompleted() { } @Override public void onError(Throwable e) { } @Override public void onNext(String s) { } //使用Single,Observable还有一个精简版 sub = Single.fromCallable(new Callable<String>() { @Override public String call() throws Exception { return null; } }).subscribeOn(Schedulers.newThread()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new SingleSubscriber<String>(){ @Override public void onSuccess(String value) { } @Override public void onError(Throwable error) { } });

9.Action1：RxJava中的一个接口，封装了含有 1 个参的call（）方法，即call（T t）。

10.Func1:和Action1类似，但是有返回值，Action1没有返回值

Observable.create(new Observable.OnSubscribe<Student>() { @Override public void call(Subscriber<? super Student> subscriber) { subscriber.onNext(getStudentInfo(123456777)); subscriber.onCompleted(); } }).map(new Func1<Student, String>() { @Override public String call(Student student) { return student.getName(); } }) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Action1<String>() { @Override public void call(String s) { mTvMsg.setText(s); } });