RxJava初探

Posted by 程序亦非猿 on 2015-09-11

RxJava

RxJava – Reactive Extensions for the JVM – a library for composing asynchronous and event-based programs using observable sequences for the Java VM.

RxJava 是java的响应式编程的扩展.

Ps:RxJava 配 Lambda 更合适哦!~
不过这里没用Lambda,本文只关心RxJava语法,因为现在是学好基础语法,打好基础的时候,如果你还不知道Lambda的可以看看我的使用Lambda在Android中的替代匿名类

依赖:compile 'io.reactivex:rxjava:1.0.12'

基础知识

RxJava最核心的两个东西:

  1. Observables(被观察者,事件源)
  2. Subscribers(观察者,用来响应Observables发出的事件)。

Observables发出一系列事件,Subscribers处理这些事件。
这里的事件可以是任何你感兴趣的东西 (触摸事件,web接口调用返回的数据。。。)

一个Observable可以发出零个或者多个事件,知道结束或者出错。每发出一个事件,就会调用它的Subscriber的onNext方法,最后调用Subscriber.onNext()或者Subscriber.onError()结束。

粗看跟观察者模式很相似,但是有一点明显不同,那就是:
如果一个Observerble没有任何的的Subscriber,那么这个Observable是不会发出任何事件的。

Talk is cheap ,show me the code.
接下去练习一下~

练习

1. 创建Observable
通过Observable.create方法很容易创建

1
2
3
4
5
6
7
8
9
10
static Observable<String> mStringObservable = Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
subscriber.onNext("Hello Rxjava");
//Notice 如果onCompleted 先调用了,那么后面的onNext onError则会被无视掉,但是不是等同于return,后面其他的代码还是会执行
subscriber.onCompleted();
System.out.println("看看onCompleted调用了还会不会有输出");
// subscriber.onError(new Throwable("hehehe"));
}
});

2. 创建Subscriber

接着我们创建一个Subscriber来处理Observable对象发出的字符串。

很简单,只是打印一下数据而已.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
static Subscriber<String> mStringSubscriber = new Subscriber<String>() {
@Override
public void onCompleted() {
System.out.println("onCompleted");
}
@Override
public void onError(Throwable e) {
System.out.println("onError:" + e.getMessage());
}
@Override
public void onNext(String s) {
System.out.println(s);
}
};

3. 订阅

main方法添加代码,run起来~

1
2
3
public static void main(String[] args) {
mStringObservable.subscribe(mStringSubscriber);
}

输出Log:

1
2
3
Hello Rxjava
onCompleted
看看onCompleted调用了还会不会有输出

注意点:

  1. 如果调用了subscriber.onCompleted(),那么在onCompleted后调用的onNextonError都不会生效.

其实如果Rx真的每次都需要这么写的话,真的是比Java还啰嗦,接下去练习更简洁的方法.

简化代码

Observable.just方法可以用来创建只发出一个事件就结束的Observable对象
上面创建Observable对象的代码可以简化为一行

1
static Observable<String> simpleObservable = Observable.just("Hell,simple observable");

跟踪一下just方法的代码到ScalarSynchronousObservable类,会发现,其实它就是创建了一个OnSubscribe对象,顺便帮忙调用了onNextonCompleted:

1
2
3
4
5
6
7
8
9
10
protected ScalarSynchronousObservable(final T t) {
super(new OnSubscribe<T>() {
@Override
public void call(Subscriber<? super T> s) {
s.onNext(t);
s.onCompleted();
}
});
this.t = t;
}

同样,订阅一下试试~

1
simpleObservable.subscribe(mStringSubscriber);

输出Log:

1
2
Hell,simple observable
onCompleted

更进一步

RxJava中还有Action1Action0这两个类.

另外
subscribe方法有一个重载版本,接受三个Action类型的参数,分别对应OnNext,OnComplete, OnError函数。

subscribe(onNextAction, onErrorAction, onCompleteAction);

如果我们并不关心OnCompleteOnError,那么完全可以用Action1来替代上面所写:

1
2
3
4
5
6
static Action1<String> onNextAction = new Action1<String>() {
@Override
public void call(String s) {
System.out.println("Action1<String> onNextAction:" + s);
}
};

在main中增加:

1
mStringObservable.subscribe(onNextAction);

run起来~
Log:

1
Action1<String> onNextAction:Hello Rxjava

那么再写一下onComplete和onError看:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
// 注意是:Throwable
static Action1<Throwable> onErrorAction = new Action1<Throwable>() {
@Override
public void call(Throwable throwable) {
System.out.println("Action1<Throwable> onErrorAction:");
}
};
//Action0
static Action0 onCompleteAction = new Action0() {
@Override
public void call() {
System.out.println("Action0 onCompleteAction:");
}
};

1
2
3
4
//Notice 如果调用onError 这个单个的onNextAction 会报错
mStringObservable.subscribe(onNextAction);
mStringObservable.subscribe(onNextAction, onErrorAction);
mStringObservable.subscribe(onNextAction, onErrorAction, onCompleteAction);

需要注意的是onErrorAction 接收的是Throwable,onComplete无类型
另外如果call里面调用了onError则会报错:
Exception in thread "main" rx.exceptions.OnErrorNotImplementedException: hehehe

再简化,单个Action1可以简化成:

1
2
3
4
5
6
Observable.just("Hello simple onaction1").subscribe(new Action1<String>() {
@Override
public void call(String s) {
System.out.println(s);
}
});

输出日志:

1
Hello simple onaction1

好了,基本语法部分例子学完了,例子很简单,到此结束,下一篇Rx操作符.

参考/更多

Dan Lew大神Rx系列
大头鬼整理