RxJava基本使用

更多文章请点击连接：http://77blogs.com/?p=162

转载请标明出处：http://www.javashuo.com/article/p-pfcoiphs-mh.html，http://77blogs.com/?p=162

 

RxJava到底是啥，从根本上来说，它就是一个实现异步操做的库，而且可以使代码很是简洁。它的异步是使用观察者模式来实现的。

关于观察者模式的介绍，能够看我这一篇文章：

http://www.javashuo.com/article/p-bnuyxkop-mg.html 

这里我主要讲RxJava的一些基本用法，基本案例，原理的话暂时不深究：

1、本身构造事件

 

Observable.create(new ObservableOnSubscribe<Integer>() { @Override public void subscribe(ObservableEmitter emitter) { int i = getNumber(); if (i < 0) { emitter.onComplete(); return; } else { Log.d(TAG, Thread.currentThread().getName()); emitter.onNext(i); emitter.onComplete(); } } }) .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(new Consumer<Integer>() { @Override public void accept(Integer integer) throws Exception { Log.d(TAG, Thread.currentThread().getName()); Log.d(TAG, integer + ""); } }, new Consumer<Throwable>() { @Override public void accept(Throwable throwable) throws Exception { } });

RxJava 有四个基本概念：Observable (可观察者，即被观察者)、 Observer (观察者)、 subscribe (订阅)、事件。Observable 和 Observer 经过 subscribe() 方法实现订阅关系，从而 Observable 能够在须要的时候发出事件来通知 Observer。

 

onNext():方法用来发送事件。

下面看看其余两个方法：

• onCompleted(): 事件队列完结。RxJava 不只把每一个事件单独处理，还会把它们看作一个队列。RxJava 规定，当不会再有新的 onNext() 发出时，须要触发 onCompleted() 方法做为标志。
• onError(): 事件队列异常。在事件处理过程当中出异常时，onError() 会被触发，同时队列自动终止，不容许再有事件发出。
• 在一个正确运行的事件序列中, onCompleted() 和 onError() 有且只有一个，而且是事件序列中的最后一个。须要注意的是，onCompleted() 和 onError() 两者也是互斥的，即在队列中调用了其中一个，就不该该再调用另外一个。

讲一下咱们上面的例子，上面这个例子是采用简洁的链式调用来写的：

首先使用 create() 方法来建立一个 Observable ，并为它定义事件触发规则，而后经过emitter.onNext(i)传递出来，.subscribeOn(Schedulers.io())即是指定该事件产生的所在的线程为子线程，.observeOn(AndroidSchedulers.mainThread())指定观察者执行的线程为主线程。这时候为止返回的对象为Observable对象。

而后该Observable对象subscribe绑定观察者（也就是观察者进行订阅），里面有接收被观察者发出来的事件，有一个成功的方法，和一个失败的方法，这样就实现了由被观察者向观察传递事件。

 

2、对集合里的数据进行变换

        List<Integer> list = new ArrayList<Integer>() { { add(0); add(1); add(2); } }; Observable.fromIterable(list).map(new Function() { @Override public Object apply(Object o) throws Exception { int i = (int) o + 1; return String.valueOf(i); } }) .toList() .toObservable().subscribeOn(Schedulers.io()) .subscribeOn(AndroidSchedulers.mainThread()) .subscribe(new Consumer() { @Override public void accept(Object o) throws Exception { Log.d(TAG, o.toString()); } });

且看，咱们须要对某个集合里面的数据一一进行变换，而后发送出来执行其余操做。

上面即是对集合里面的每一项进行加一操做，而后再转换为String类型，而后toList()，组合成集合发送出来，最后在观察者方法中打印出每一项。

 

3、合并执行

定义两个被观察者，各自产生事件，而后合并在一块儿，发送给一个观察者。

首先定义咱们上面第一个例子的被观察者，用于发送一个数字：

        Observable observable1 = Observable.create(new ObservableOnSubscribe<Integer>() { @Override public void subscribe(ObservableEmitter emitter) { int i = getNumber(); if (i < 0) { emitter.onComplete(); return; } else { Log.d(TAG, Thread.currentThread().getName()); emitter.onNext(i); emitter.onComplete(); } } }) .subscribeOn(Schedulers.io());

 

其次再定义咱们上面第二个例子的被观察者：

        List<Integer> list = new ArrayList<Integer>() { { add(0); add(1); add(2); } }; Observable observable2 = Observable.fromIterable(list).map(new Function() { @Override public Object apply(Object o) { int i = (int) o + 1; return String.valueOf(i); } }) .toList() .toObservable().subscribeOn(Schedulers.io());

 

最后将这两个被观察者的事件合并起来发送给一个观察者：

        Disposable disposable = Observable.zip(observable1, observable2, new BiFunction() { @Override public Object apply(Object o, Object o2) throws Exception { int i = (int) o; String k = (String) ((List) o2).get(0); return k + i; } }) .subscribe(new Consumer() { @Override public void accept(Object o) { Log.d(TAG, (String) o); } }, new Consumer<Throwable>() { @Override public void accept(Throwable throwable) { Log.d(TAG, throwable.getMessage()); } });

zip方法，顾名思义，有点相似与于打包的意思。

o为被观察者1返回的结果，o2为被观察2返回的结果，将这两个结果一块儿处理后发送给观察者。打印出来。

如今先介绍这几个，找个时间再整理一些其余的用法以及原理实现。