最基本的调用
<code>Observable.create(new ObservableOnSubscribe<string>() {
@Override
public void subscribe(ObservableEmitter<string> emitter) throws Exception {
emitter.onNext("1");
emitter.onNext("2");
emitter.onNext("3");
emitter.onNext("4");
emitter.onNext("5");
emitter.onComplete();
XLog.d("subscribe over");
}
}).subscribe(new Observer<string>() {
@Override
public void onSubscribe(Disposable d) {
XLog.d("onSubscribe ");
}
@Override
public void onNext(String s) {
XLog.d("onNext " + s);
}
@Override
public void onError(Throwable e) {
XLog.d("onError ");
}
@Override
public void onComplete() {
XLog.d("onComplete ");
}
});
/<string>/<string>/<string>/<code>
输出结果：基本按照顺序输出
onSubscribeonNext 1onNext 2onNext 3onNext 4onNext 5onCompletesubscribe over
onComplete/onError放中间
<code>emitter.onNext("1");
emitter.onNext("2");
emitter.onNext("3");
//emitter.onComplete();
emitter.onError(new Exception());
emitter.onNext("4");
emitter.onNext("5");
/<code>
输出结果：当中间有 onCompelete/onError 时，Observable会继续发送，但Obsever不会再接收事件了
onSubscribeonNext 1onNext 2onNext 3//onCompleteonErrorsubscribe over
连续两次调用onComplete
<code> emitter.onNext("1");
emitter.onNext("2");
emitter.onNext("3");
emitter.onComplete();
emitter.onComplete();
emitter.onNext("4");
emitter.onNext("5");
/<code>
输出结果：和单次调用 onComplete一样，如果放中间，Observer将不会再接收事件，且无异常
onSubscribeonNext 1onNext 2onNext 3onCompletesubscribe over
连续两次调用onError
<code> emitter.onNext("1");
emitter.onNext("2");
emitter.onNext("3");
emitter.onError(new Exception());
emitter.onError(new Exception());
emitter.onComplete();
emitter.onNext("4");
emitter.onNext("5");
XLog.d("subscribe over");
/<code>
输出结果：收到onError事件便截止
onSubscribeonNext 1onNext 2onNext 3onError
除此之外，连续 2 次调用 onError(),还报出了一个 UndeliverableException
<code>09-22 12:45:18.181 8830-8830/www.ly.com.rxjavademo W/System.err: io.reactivex.exceptions.UndeliverableException: java.lang.Exception
at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:367)
at io.reactivex.internal.operators.observable.ObservableCreate$CreateEmitter.onError(ObservableCreate.java:74)
at www.ly.com.rxjavademo.MainActivity$2.subscribe(MainActivity.java:73)
at io.reactivex.internal.operators.observable.ObservableCreate.subscribeActual(ObservableCreate.java:40)
at io.reactivex.Observable.subscribe(Observable.java:12005)
at www.ly.com.rxjavademo.MainActivity.onCreate(MainActivity.java:81)
/<code>
onError 和 onComplete 同时出现
<code> emitter.onNext("1");
emitter.onNext("2");
emitter.onNext("3");
emitter.onComplete();
emitter.onError(new Exception());
emitter.onNext("4");
emitter.onNext("5");
/<code>
输出结果：输出到onComplete截止
onSubscribeonNext 1onNext 2onNext 3onComplete
除此之外，同时也收到了 UndeliverableException 而终止进程
<code>09-22 13:10:32.738 14353-14353/www.ly.com.rxjavademo W/System.err: io.reactivex.exceptions.UndeliverableException: java.lang.Exception
at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:367)
09-22 13:10:32.739 14353-14353/www.ly.com.rxjavademo W/System.err: at io.reactivex.internal.operators.observable.ObservableCreate$CreateEmitter.onError(ObservableCreate.java:74)
at www.ly.com.rxjavademo.MainActivity$2.subscribe(MainActivity.java:73)
at io.reactivex.internal.operators.observable.ObservableCreate.subscribeActual(ObservableCreate.java:40)
at io.reactivex.Observable.subscribe(Observable.java:12005)
at www.ly.com.rxjavademo.MainActivity.onCreate(MainActivity.java:81)
/<code>
但如果把 onError 放在 onComplete 前面呢？则输出结果为：
onSubscribeonNext 1onNext 2onNext 3onErrorsubscribe over
说明onError 之前不能有 onComplete / onError 事件发生。
指定调度线程
<code>Observable.create(new ObservableOnSubscribe<string>() {
@Override
public void subscribe(ObservableEmitter<string> emitter) throws Exception {
emitter.onNext("1");
emitter.onNext("2");
emitter.onNext("3");
emitter.onNext("4");
emitter.onNext("5");
XLog.d("subscribe over");
}
})
.subscribeOn(Schedulers.newThread())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<string>() {
@Override
public void onSubscribe(Disposable d) {
XLog.d("onSubscribe ");
}
@Override
public void onNext(String s) {
XLog.d("onNext " + s);
}
@Override
public void onError(Throwable e) {
XLog.d("onError ");
}
@Override
public void onComplete() {
XLog.d("onComplete ");
}
});
/<string>/<string>/<string>/<code>
输出结果：被观察者在新的线程中执行发送事件，观察者在Android主线程中接收事件
Thread: mainonSubscribeThread: RxNewThreadScheduler-1subscribe overThread: mainonNext 1Thread: mainonNext 2Thread: mainonNext 3Thread: mainonNext 4Thread: mainonNext 5
压力测试
被观察者限制快速发送数据，观察察者每休息 5 秒钟处理一个 onNext 事件
<code>Observable.create(new ObservableOnSubscribe<string>() {
@Override
public void subscribe(ObservableEmitter<string> emitter) throws Exception {
int i = 0;
for (;;) {
emitter.onNext("send " + (i++));
}
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer<string>() {
@Override
public void onSubscribe(Disposable d) {
XLog.d("onSubscribe ");
}
@Override
public void onNext(String s) {
XLog.d("onNext " + s);
try {
Thread.sleep(5000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public void onError(Throwable e) {
XLog.d("onError ");
}
@Override
public void onComplete() {
XLog.d("onComplete ");
}
});
/<string>/<string>/<string>/<code>
输出结果：内存溢出！！！在RxJava2中Observable是不支持背压的
<code>09-22 13:36:52.613 20160-20169/www.ly.com.rxjavademo W/art: Throwing OutOfMemoryError "Failed to allocate a 28 byte allocation with 1864 free bytes and 1864B until OOM; failed due to fragmentation (required continguous free 4096 bytes for a new buffer where largest contiguous free 0 bytes)" (recursive case)
/<code>
<code>09-22 13:36:52.613 20160-20169/www.ly.com.rxjavademo W/art: Throwing OutOfMemoryError "Failed to allocate a 28 byte allocation with 1864 free bytes and 1864B until OOM; failed due to fragmentation (required continguous free 4096 bytes for a new buffer where largest contiguous free 0 bytes)" (recursive case)
/<code>