RxJava 响应式编程
Rxjava2入门教程五：Flowable背压支持——对Flowable最全面而详细的讲解
public class DemoActivity extends AppCompatActivity {
private final int count = 60;
private TextView textView;
private Button button;
@Override
protected void onCreate(Bundle savedInstanceState) {
super.onCreate(savedInstanceState);
setContentView(R.layout.activity_demo);
textView = findViewById(R.id.textView);
button = findViewById(R.id.button);
button.setOnClickListener(new View.OnClickListener() {
@Override
public void onClick(View v) {
testBackpressure24();
}
});
}
private Subscription mSubscription;
public void testBackpressure24() {
Flowable.create(new FlowableOnSubscribe() {
@Override
public void subscribe(FlowableEmitter emitter) throws Exception {
int n = 0;
while (true){
if (emitter.requested() == 0) continue;
emitter.onNext(n);
System.out.println("发射---->" + n);
n++;
}
}
}, BackpressureStrategy.MISSING)
.subscribeOn(Schedulers.newThread())
.observeOn(Schedulers.newThread())
.subscribe(new FlowableSubscriber() {
@Override
public void onSubscribe(Subscription s) {
System.out.println("onSubscribe----");
s.request(1);
mSubscription = s;
}
@Override
public void onNext(Integer integer) {
try {
Thread.sleep(50L);
System.out.println("接收---->" + integer);
mSubscription.request(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public void onError(Throwable t) {
System.out.println("onError----");
t.printStackTrace();
}
@Override
public void onComplete() {
System.out.println("onComplete----");
}
});
}
public void testBackpressure23() {
Flowable.create(new FlowableOnSubscribe() {
@Override
public void subscribe(FlowableEmitter emitter) throws Exception {
String threadName = Thread.currentThread().getName();
System.out.println(threadName + "开始发射数据" + System.currentTimeMillis());
for (int i = 1; i <= 500; i++) {
System.out.println(threadName + "发射---->" + i);
emitter.onNext(i);
try {
Thread.sleep(100);//每隔100毫秒发射一次数据
} catch (Exception ex) {
emitter.onError(ex);
}
}
System.out.println(threadName + "发射数据结束" + System.currentTimeMillis());
emitter.onComplete();
}
}, BackpressureStrategy.MISSING)
.onBackpressureLatest()
.subscribeOn(Schedulers.newThread())
.observeOn(Schedulers.newThread())
.subscribe(new Subscriber() {
@Override
public void onSubscribe(Subscription s) {
s.request(Integer.MAX_VALUE);
}
@Override
public void onNext(Integer integer) {
try {
Thread.sleep(300);//每隔300毫秒接收一次数据
} catch (InterruptedException ignore) {
}
System.out.println(Thread.currentThread().getName() + "接收---------->" + integer);
}
@Override
public void onError(Throwable t) {
t.printStackTrace();
}
@Override
public void onComplete() {
System.out.println(Thread.currentThread().getName() + "接收----> 完成");
}
});
}
public void testBackpressure22() {
Flowable.create(new FlowableOnSubscribe() {
@Override
public void subscribe(FlowableEmitter emitter) throws Exception {
String threadName = Thread.currentThread().getName();
System.out.println(threadName + "开始发射数据" + System.currentTimeMillis());
for (int i = 1; i <= 500; i++) {
System.out.println(threadName + "发射---->" + i);
emitter.onNext(i);
try {
Thread.sleep(100);//每隔100毫秒发射一次数据
} catch (Exception ex) {
emitter.onError(ex);
}
}
System.out.println(threadName + "发射数据结束" + System.currentTimeMillis());
emitter.onComplete();
}
}, BackpressureStrategy.LATEST)
.subscribeOn(Schedulers.newThread())
.observeOn(Schedulers.newThread())
.subscribe(new Subscriber() {
@Override
public void onSubscribe(Subscription s) {
s.request(Integer.MAX_VALUE);
}
@Override
public void onNext(Integer integer) {
try {
Thread.sleep(300);//每隔300毫秒接收一次数据
} catch (InterruptedException ignore) {
}
System.out.println(Thread.currentThread().getName() + "接收---------->" + integer);
}
@Override
public void onError(Throwable t) {
t.printStackTrace();
}
@Override
public void onComplete() {
System.out.println(Thread.currentThread().getName() + "接收----> 完成");
}
});
}
public void testBackpressure21() {
Flowable.create(new FlowableOnSubscribe() {
@Override
public void subscribe(FlowableEmitter e) throws Exception {
for (int i = 1; i <= 129; i++) { // 默认缓存大小是128
e.onNext(i);
}
e.onComplete();
}
}, BackpressureStrategy.ERROR)
.subscribeOn(Schedulers.newThread())
.observeOn(Schedulers.newThread())
.subscribe(new Subscriber() {
@Override
public void onSubscribe(Subscription s) {
s.request(Long.MAX_VALUE); //注意此处，暂时先这么设置
}
@Override
public void onNext(Integer integer) {
try {
Thread.sleep(10000);
} catch (InterruptedException ignore) {
}
System.out.println(integer);
}
@Override
public void onError(Throwable t) {
t.printStackTrace();
}
@Override
public void onComplete() {
System.out.println("接收----> 完成");
}
});
}
public void testOnBackpressure2() {
Flowable.range(1, 500)
// .onBackpressureLatest()
.onBackpressureDrop()
.subscribeOn(Schedulers.newThread())
.observeOn(Schedulers.newThread())
.subscribe(new Subscriber() {
@Override
public void onSubscribe(Subscription s) {
s.request(Integer.MAX_VALUE);
}
@Override
public void onNext(Integer integer) {
try {
Thread.sleep(100);//每隔300毫秒接收一次数据
} catch (InterruptedException ignore) {
}
System.out.println(Thread.currentThread().getName() + "接收---------->" + integer);
}
@Override
public void onError(Throwable t) {
t.printStackTrace();
}
@Override
public void onComplete() {
System.out.println(Thread.currentThread().getName() + "接收----> 完成");
}
});
}
private void testOnBackpressure1() {
Flowable.range(1, 500)
// .map(new Function() {
// @Override
// public Integer apply(Integer integer) throws Exception {
// Log.e("TAG", "发射：" + integer);
// return integer;
// }
// })
.onBackpressureBuffer(5, new Action() {
@Override
public void run() throws Exception {
Log.e("TAG", "溢出");
}
}, BackpressureOverflowStrategy.DROP_OLDEST)
.observeOn(Schedulers.computation())
.subscribe(new FlowableSubscriber() {
@Override
public void onSubscribe(Subscription s) {
Log.e("TAG", "onSubscribe");
s.request(Integer.MAX_VALUE);
}
@Override
public void onNext(Integer integer) {
try {
Thread.sleep(100);
Log.e("TAG", "接收" + integer);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public void onError(Throwable t) {
Log.e("TAG", "onError");
t.printStackTrace();
}
@Override
public void onComplete() {
Log.e("TAG", "onComplete");
}
});
}
private void testPublish() {
PublishProcessor processor = PublishProcessor.create();
processor
.sample(1, TimeUnit.SECONDS)
.observeOn(Schedulers.computation(), false, 1024)
.subscribe(new FlowableSubscriber() {
@Override
public void onSubscribe(Subscription s) {
Log.e("TAG", "onSubscribe");
s.request(Integer.MAX_VALUE);
}
@Override
public void onNext(Integer integer) {
try {
Thread.sleep(500);
Log.e("TAG", "onNext" + integer);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public void onError(Throwable t) {
Log.e("TAG", "onError");
t.printStackTrace();
}
@Override
public void onComplete() {
Log.e("TAG", "onComplete");
}
});
for (int i = 0; i < 1_000; i++) {
processor.onNext(i);
}
}
private void testFlowable() {
Flowable.range(1, Integer.MAX_VALUE)
.observeOn(Schedulers.computation())
.subscribe(new FlowableSubscriber() {
@Override
public void onSubscribe(Subscription s) {
Log.e("TAG", "onSubscribe");
s.request(Integer.MAX_VALUE);
}
@Override
public void onNext(Integer integer) {
try {
Thread.sleep(1000);
Log.e("TAG", "onNext" + integer);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
@Override
public void onError(Throwable t) {
Log.e("TAG", "onError");
t.printStackTrace();
}
@Override
public void onComplete() {
Log.e("TAG", "onComplete");
}
});
}
private void testTo() {
Observable.range(1, 10)
.to(new Function, Single>() {
@Override
public Single apply(Observable integerObservable) throws Exception {
return integerObservable.count();
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new SingleObserver() {
@Override
public void onSubscribe(Disposable d) {
Log.e("TAG", "onSubscribe");
}
@Override
public void onSuccess(Long aBoolean) {
Log.e("TAG", "onSuccess: " + aBoolean);
textView.setText(String.valueOf(aBoolean));
}
@Override
public void onError(Throwable e) {
Log.e("TAG", "onError");
e.printStackTrace();
}
});
}
private void testToMap() {
Observable.just(1, 2, 3, 4, 5, 6)
.toMap(Integer::toHexString)
// .toMultimap(Integer::toHexString)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new SingleObserver>() {
@Override
public void onSubscribe(Disposable d) {
Log.e("TAG", "onSubscribe");
}
@Override
public void onSuccess(Map aBoolean) {
Log.e("TAG", "onSuccess: " + aBoolean);
textView.setText(String.valueOf(aBoolean));
}
@Override
public void onError(Throwable e) {
Log.e("TAG", "onError");
e.printStackTrace();
}
});
}
private void testDefaultEmpty() {
Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter emitter) throws Exception {
emitter.onComplete();
}
}).defaultIfEmpty(6)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer() {
@Override
public void onSubscribe(Disposable d) {
Log.e("TAG", "onSubscribe");
}
@Override
public void onNext(Integer aBoolean) {
Log.e("TAG", "onSuccess: " + aBoolean);
textView.setText(String.valueOf(aBoolean));
}
@Override
public void onError(Throwable e) {
Log.e("TAG", "onError");
e.printStackTrace();
}
@Override
public void onComplete() {
Log.e("TAG", "onComplete");
}
});
}
private void testAmb() {
Observable.amb(
Arrays.asList(Observable.range(1, 5), Observable.range(6, 5))
).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer() {
@Override
public void onSubscribe(Disposable d) {
Log.e("TAG", "onSubscribe");
}
@Override
public void onNext(Integer aBoolean) {
Log.e("TAG", "onSuccess: " + aBoolean);
textView.setText(String.valueOf(aBoolean));
}
@Override
public void onError(Throwable e) {
Log.e("TAG", "onError");
e.printStackTrace();
}
@Override
public void onComplete() {
Log.e("TAG", "onComplete");
}
});
}
private void testSequenceEqual() {
Observable.sequenceEqual(
Observable.range(1, 5),
Observable.range(1, 6)
).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new SingleObserver() {
@Override
public void onSubscribe(Disposable d) {
Log.e("TAG", "onSubscribe");
}
@Override
public void onSuccess(Boolean aBoolean) {
Log.e("TAG", "onSuccess: " + aBoolean);
textView.setText(String.valueOf(aBoolean));
}
@Override
public void onError(Throwable e) {
Log.e("TAG", "onError");
e.printStackTrace();
}
});
}
private void testAllAny() {
// all & any
// contains & isEmpty
Observable.range(5, 5)
// .all(new Predicate() {
// @Override
// public boolean test(Integer integer) throws Exception {
// return integer >= 5; // true
// }
// })
// .any(new Predicate() {
// @Override
// public boolean test(Integer integer) throws Exception {
// return integer > 5; // true
// }
// })
// .contains(4) // false
.isEmpty() // false
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new SingleObserver() {
@Override
public void onSubscribe(Disposable d) {
Log.e("TAG", "onSubscribe");
}
@Override
public void onSuccess(Boolean aBoolean) {
Log.e("TAG", "onSuccess: " + aBoolean);
textView.setText(String.valueOf(aBoolean));
}
@Override
public void onError(Throwable e) {
Log.e("TAG", "onError");
e.printStackTrace();
}
});
}
private void testRetry() {
Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter emitter) throws Exception {
emitter.onNext(1);
emitter.onError(new Throwable("Error 1"));
// emitter.onError(new Throwable("Error 2"));
emitter.onNext(2);
}
})
.retry(2)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer() {
@Override
public void onSubscribe(Disposable d) {
Log.e("TAG", "onSubscribe");
}
@Override
public void onNext(Integer integer) {
Log.e("TAG", "onNext: " + integer);
textView.setText(String.valueOf(integer));
}
@Override
public void onError(Throwable e) {
Log.e("TAG", "onError");
e.printStackTrace();
}
@Override
public void onComplete() {
Log.e("TAG", "onComplete");
}
});
}
private void testCatch() {
Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter emitter) throws Exception {
emitter.onNext(1);
emitter.onError(null);
emitter.onNext(2);
}
})
// .onErrorReturn(new Function() {
// @Override
// public Integer apply(Throwable throwable) throws Exception {
// return 666;
// }
// })
.onErrorResumeNext(Observable.just(11, 22, 33))
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer() {
@Override
public void onSubscribe(Disposable d) {
Log.e("TAG", "onSubscribe");
// Log.e("TAG", "thread: " + Thread.currentThread().getName());
}
@Override
public void onNext(Integer integer) {
Log.e("TAG", "onNext: " + integer);
// Log.e("TAG", "thread: " + Thread.currentThread().getName());
textView.setText(String.valueOf(integer));
}
@Override
public void onError(Throwable e) {
Log.e("TAG", "onError");
e.printStackTrace();
}
@Override
public void onComplete() {
Log.e("TAG", "onComplete");
}
});
}
private void testCombineLatest() {
Observable.combineLatest(
Observable.range(1, 6).subscribeOn(Schedulers.io()),
Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter emitter) throws Exception {
emitter.onNext("a");
emitter.onNext("b");
emitter.onNext("c");
emitter.onNext("d");
}
}).subscribeOn(Schedulers.io()),
new BiFunction() {
@Override
public String apply(Integer integer, String str) throws Exception {
return integer + "==" + str;
}
}
).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer() {
@Override
public void onSubscribe(Disposable d) {
Log.e("TAG", "onSubscribe");
// Log.e("TAG", "thread: " + Thread.currentThread().getName());
}
@Override
public void onNext(String integer) {
Log.e("TAG", "onNext: " + integer);
// Log.e("TAG", "thread: " + Thread.currentThread().getName());
textView.setText(String.valueOf(integer));
}
@Override
public void onError(Throwable e) {
Log.e("TAG", "onError");
e.printStackTrace();
}
@Override
public void onComplete() {
Log.e("TAG", "onComplete");
}
});
}
private void testZip() {
Observable.zip(
Observable.range(1, 6).subscribeOn(Schedulers.io()),
Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter emitter) throws Exception {
emitter.onNext("a");
emitter.onNext("b");
emitter.onNext("c");
emitter.onNext("d");
}
}).subscribeOn(Schedulers.io()),
new BiFunction() {
@Override
public String apply(Integer integer, String str) throws Exception {
return integer + "==" + str;
}
}
).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer() {
@Override
public void onSubscribe(Disposable d) {
Log.e("TAG", "onSubscribe");
// Log.e("TAG", "thread: " + Thread.currentThread().getName());
}
@Override
public void onNext(String integer) {
Log.e("TAG", "onNext: " + integer);
// Log.e("TAG", "thread: " + Thread.currentThread().getName());
textView.setText(String.valueOf(integer));
}
@Override
public void onError(Throwable e) {
Log.e("TAG", "onError");
e.printStackTrace();
}
@Override
public void onComplete() {
Log.e("TAG", "onComplete");
}
});
}
private void testConcat() {
Observable.concat(
Observable.range(1, 20).subscribeOn(Schedulers.io()),
Observable.range(100, 20).subscribeOn(Schedulers.io())
// Observable.create((ObservableOnSubscribe) observableEmitter -> {
// Thread.sleep(50);
// observableEmitter.onError(new Exception("error"));
// }).subscribeOn(Schedulers.io())
).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer() {
@Override
public void onSubscribe(Disposable d) {
Log.e("TAG", "onSubscribe");
// Log.e("TAG", "thread: " + Thread.currentThread().getName());
}
@Override
public void onNext(Integer integer) {
Log.e("TAG", "onNext: " + integer);
// Log.e("TAG", "thread: " + Thread.currentThread().getName());
textView.setText(String.valueOf(integer));
}
@Override
public void onError(Throwable e) {
Log.e("TAG", "onError");
e.printStackTrace();
}
@Override
public void onComplete() {
Log.e("TAG", "onComplete");
}
});
}
private void testMerge() {
Observable.merge(
Observable.range(1, 20).subscribeOn(Schedulers.io()),
Observable.range(100, 20).subscribeOn(Schedulers.io())
// Observable.create((ObservableOnSubscribe) observableEmitter -> {
// Thread.sleep(50);
// observableEmitter.onError(new Exception("error"));
// }).subscribeOn(Schedulers.io())
).subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer() {
@Override
public void onSubscribe(Disposable d) {
Log.e("TAG", "onSubscribe");
// Log.e("TAG", "thread: " + Thread.currentThread().getName());
}
@Override
public void onNext(Integer integer) {
Log.e("TAG", "onNext: " + integer);
// Log.e("TAG", "thread: " + Thread.currentThread().getName());
textView.setText(String.valueOf(integer));
}
@Override
public void onError(Throwable e) {
Log.e("TAG", "onError");
e.printStackTrace();
}
@Override
public void onComplete() {
Log.e("TAG", "onComplete");
}
});
}
private void testDebounce() {
Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter emitter) throws InterruptedException {
emitter.onNext(1);
Thread.sleep(200);
emitter.onNext(2);
// 500 :输出2
Thread.sleep(800);
// 1000：无输出
emitter.onNext(3);
Thread.sleep(200);
emitter.onNext(4);
// 1500：输出4
Thread.sleep(800);
emitter.onNext(5);
// 2000：输出5
Thread.sleep(800);
emitter.onNext(6);
// 2500：输出6
Thread.sleep(100);
emitter.onComplete();
}
}).debounce(500, TimeUnit.MILLISECONDS)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer() {
@Override
public void onSubscribe(Disposable d) {
Log.e("TAG", "onSubscribe");
// Log.e("TAG", "thread: " + Thread.currentThread().getName());
}
@Override
public void onNext(Integer integer) {
Log.e("TAG", "onNext: " + integer);
// Log.e("TAG", "thread: " + Thread.currentThread().getName());
textView.setText(String.valueOf(integer));
}
@Override
public void onError(Throwable e) {
Log.e("TAG", "onError");
e.printStackTrace();
}
@Override
public void onComplete() {
Log.e("TAG", "onComplete");
}
});
}
private void testFilter() {
Observable.range(2, 5)
.elementAt(0)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new MaybeObserver() {
@Override
public void onSubscribe(Disposable d) {
Log.e("TAG", "onSubscribe");
// Log.e("TAG", "thread: " + Thread.currentThread().getName());
}
@Override
public void onSuccess(Integer integer) {
Log.e("TAG", "onSuccess: " + integer);
// Log.e("TAG", "thread: " + Thread.currentThread().getName());
textView.setText(String.valueOf(integer));
}
@Override
public void onError(Throwable e) {
Log.e("TAG", "onError");
e.printStackTrace();
}
@Override
public void onComplete() {
Log.e("TAG", "onComplete");
}
});
}
private void testScan() {
Observable.range(2, 5)
.scan(new BiFunction() {
@Override
public Integer apply(Integer integer, Integer integer2) throws Exception {
Log.e("TAG", "integer=" + integer + "==" + integer2);
return integer * integer2;
}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Observer() {
@Override
public void onSubscribe(Disposable d) {
Log.e("TAG", "onSubscribe");
// Log.e("TAG", "thread: " + Thread.currentThread().getName());
}
@Override
public void onNext(Integer s) {
Log.e("TAG", "onNext: " + s);
// Log.e("TAG", "thread: " + Thread.currentThread().getName());
textView.setText(String.valueOf(s));
}
@Override
public void onError(Throwable e) {
Log.e("TAG", "onError");
e.printStackTrace();
}
@Override
public void onComplete() {
Log.e("TAG", "onComplete");
}
});
}
private void testGroupby() {
ArrayList 
