RxJava的坑

情况一：

Observable.create((Observable.OnSubscribe<String>) subscriber -> {
subscriber.onNext("aaaa");
subscriber.onNext("bbbb");
subscriber.onError(new NullPointerException());
subscriber.onNext("cccc");
subscriber.onNext("dddd");
subscriber.onCompleted();
}).subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
System.out.println("onCompleted");
}

@Override
public void onError(Throwable e) {
e.printStackTrace();
}

@Override
public void onNext(String s) {
System.out.println("s=" + s);
}
});
}

这种情况输出为

s=aaaa
s=bbbb
java.lang.NullPointerException

情况二：

Observable.create((Observable.OnSubscribe<String>) subscriber -> {
subscriber.onNext("aaaa");
subscriber.onNext("bbbb");
subscriber.onError(new NullPointerException());
subscriber.onNext("cccc");
subscriber.onNext("dddd");
subscriber.onCompleted();
}).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
System.out.println("onCompleted");
}

@Override
public void onError(Throwable e) {
e.printStackTrace();
}

@Override
public void onNext(String s) {
System.out.println("s=" + s);
}
});
}

这种情况就只会输出

java.lang.NullPointerException

解决方法：

Observable.create((Observable.OnSubscribe<String>) subscriber -> {
subscriber.onNext("aaaa");
subscriber.onNext("bbbb");
subscriber.onError(new NullPointerException());
subscriber.onNext("cccc");
subscriber.onNext("dddd");
subscriber.onCompleted();
}).subscribeOn(Schedulers.io())
.materialize()
.observeOn(AndroidSchedulers.mainThread())
.<String>dematerialize()
.subscribe(new Subscriber<String>() {
@Override
public void onCompleted() {
System.out.println("onCompleted");
}

@Override
public void onError(Throwable e) {
e.printStackTrace();
}

@Override
public void onNext(String s) {
System.out.println("s=" + s);
}
});
}

Meterialize操作符将OnNext/OnError/OnComplete都转化为一个Notification对象并按照原来的顺序发射出来，而DeMeterialize则是执行相反的过程。