RXJava Observable详解 (RXJava Part
2016-12-02 本文已影响1076人
yftx_
目录更新~
RXJava Part 2 Subject
当然先看完本篇文章再看第二篇吧
使用RXJava实现观察者模式
观察者模式
简单讲，当某个对象状态发生变化时候，主动去通知相关的观察该对象的观察者。
一般使用生产者-消费者模型。
生产者：产生数据
消费者：消耗产生的数据
RXJava中的主要类
-
Observable / Subject
生产者消费者模型中的，生产者。 -
Observer / Subscriber
生产者消费者模型中的，消费者者。
RXJava对观察者模式的增强
同传统的观察者模式不同的地方为，RXJava中的Observable类增强了传统的观察者模式。
添加了新的回调分别为：
- onCompleted() 通知Observer(消费者)，生产者不会再向其传递数据了。
- onError() 通知Observer(消费者),生产者发生了错误了，不会再向其传递数据了。
RXJava消费者获取事件的方式
RXJava中消费者获取数据的方式为push模式，及生产者产生事件后，将相关事件会推送给消费者。
消费者不会主动从生产者中pull相关事件。
生产者
创建生产者的方式
- create()
可以分别传OnSubscribe，SyncOnSubscribe，AsyncOnSubscribe。
其中SyncOnSubscribe,AsyncOnSubscribe是继承OnSubscribe。
通过OnSubscribe创建的生产者，会在observable.subscribe()时发射执行OnSubscribe()中的内容。
在未通过subscribe()方法注册前，生产者中的方法不会被执行。
对于SyncOnSubscribe，AsyncOnSubscribe的介绍会放到后续文章中。
Observable<Integer> observable = Observable.create(new Observable.OnSubscribe<Integer>() {
@Override
public void call(Subscriber<? super Integer> subscriber) {
LogUtils.d(String.format("observer run in thread %s ", Thread.currentThread().getName()));
LogUtils.d("run in observable");
for (int i = 0; i < 5; i++) {
subscriber.onNext(i);
}
subscriber.onCompleted();
}
});
LogUtils.d("run after instance Observable");
observable.subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
Log.d(TAG, "event receive over");
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "producer error");
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, String.format("onNext %d ", integer.intValue()));
}
});
- from()
可以传Future/Callable,Iterable,Array.
Iterable同Array类似，会将其中的内容push给observer。
List<Integer> items = new ArrayList<>();
items.add(100);
items.add(103);
items.add(107);
Observable<Integer> observable = Observable.from(items);
LogUtils.d("run after instance Observable");
observable.subscribe(new Observer<Integer>() {
@Override
public void onCompleted() {
Log.d(TAG, "loop over");
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "error");
}
@Override
public void onNext(Integer integer) {
Log.d(TAG, String.format("onNext %d ", integer.intValue()));
}
});
Future/Callable两个类似，均为线程相关的内容。
对两者不清晰的可以参见Callable、Future和FutureTask
需要注意的是，使用future，如果future未执行完成，则会将线程卡住。
因为实际上observer会执行future.get()方法，而该方法是线程阻塞的。
所以对于生产者传递Future的场景RXJava提供了额外的参数用来设置超时时间。
详细的可以参考RXJava from的文档
ExecutorService executorService = Executors.newCachedThreadPool();
Callable<String> callable = new Callable<String>() {
@Override
public String call() throws Exception {
LogUtils.d(String.format("callable run in thread %s", Thread.currentThread().getName()));
return "run in callable call function";
}
};
FutureTask<String> task = new FutureTask<>(callable);
LogUtils.d("instance future task");
Observable<String> observable = Observable.from(task);
LogUtils.d("instance observable");
executorService.submit(task);
executorService.shutdown();
//使用observable的方式进行注册消费者，如果注册的future没有执行完成，会柱塞当前线程
observable.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
LogUtils.d("observable run complete");
}
@Override
public void onError(Throwable e) {
LogUtils.d("observable error");
}
@Override
public void onNext(String s) {
LogUtils.d(String.format("observer onNext run in thread %s ", Thread.currentThread().getName()));
LogUtils.d("observer get msg %s ", s);
}
});
- just()
just方法实际上传值的方法。可以传递1-9个值。
一般会将该方法配合项目已经有的function进行使用。
示例中的firstFunction,secondFunction,会按照just()中传参的先后顺序来执行。
无论是否observable.subscribe()被调用。其中参数传递的函数均会被执行。
因为本质上just方法是传递了一个值。
private void doObservableJust() {
Observable<String> observable = Observable.just(firstFunction(), secondFunction());
//使用just，会在创建observable时，立即执行function。
//在注册subscriber之后，会将结果发射到该subscriber中
try {
Thread.sleep(1000 * 3);
} catch (InterruptedException e) {
e.printStackTrace();
}
observable.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
Log.d(TAG, "loop over");
}
@Override
public void onError(Throwable e) {
Log.d(TAG, "error");
}
@Override
public void onNext(String s) {
Log.d(TAG, String.format("run next %s ", s));
}
});
}
private String firstFunction() {
LogUtils.d("run in firstFunction ,cur thread is %s", Thread.currentThread().toString());
return "do in some function";
}
private String secondFunction() {
LogUtils.d("run in secondFunction ,cur thread is %s", Thread.currentThread().toString());
return "do in some function";
}
- never()
一般用在测试场景，不会向observer发出任何事件。
private void doObservableNever() {
//什么都不会发送，一般用在测试场景下
Observable<String> observable = Observable.never();
observable.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
LogUtils.d("");
}
@Override
public void onError(Throwable e) {
LogUtils.d("");
}
@Override
public void onNext(String s) {
LogUtils.d("");
}
});
}
- empty()
会向observer发出结束事件，observer会调用onComplete()方法。
private void doObservableEmpty() {
//会向observer发送onComplete事件
Observable<String> observable = Observable.empty();
observable.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
LogUtils.d("");
}
@Override
public void onError(Throwable e) {
LogUtils.d("");
}
@Override
public void onNext(String s) {
LogUtils.d("");
}
});
}
- error()
会向observer发送error事件，会调用observer的onError()方法。
private void doObservableError() {
Observable<String> observable = Observable.error(new IOException());
observable.subscribe(new Observer<String>() {
@Override
public void onCompleted() {
LogUtils.d("");
}
@Override
public void onError(Throwable e) {
LogUtils.e(e.getStackTrace()+"");
}
@Override
public void onNext(String s) {
LogUtils.d("");
}
});
}
