rxjava背压(1)，4年小Android的心路历程

这样来看，我的内存就稳定，老铁稳。

那么，试试第二种方法，下游少接收点事件

//定时取样
Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter emitter) throws Exception {
for (int i = 0; ; i ) { //无限循环发事件
emitter.onNext(i);

}
}
}).sample(1, TimeUnit.SECONDS)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer() {
@Override
public void accept(Integer integer) throws Exception {

Log.d(TAG, “” integer);
}
});

或者是用过滤操作符，过滤掉一些上游事件

//过滤器 过滤操作
Observable.create(new ObservableOnSubscribe() {
@Override
public void subscribe(ObservableEmitter emitter) throws Exception {
for (int i = 0; ; i ) { //无限循环发事件
emitter.onNext(i);

}
}
}).filter(new Predicate() {
@Override
public boolean test(Integer integer) throws Exception {
return integer % 100 == 0;

}
})
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(new Consumer() {
@Override
public void accept(Integer integer) throws Exception {

Log.d(TAG, “” integer);
}
});

---

背压策略

上面唠唠叨叨说了那么多，基本上也给大家阐明了阻塞形成的原因和解决阻塞的方法，基本策略就是减少发送事件的频率和减少发送事件的数量。

But……

我们手动让上游发送事件的速度满下来貌似是不可取的，你想让上游的速度十多快呢？上游需要等多久呢？

还有……

我们依旧无法知道下游处理事件的能力，无法很好地处理阻塞的事件。

当然，你们肯定会说RxJava2.0不是很好地支了背压了吗？是的，确实比较好地对阻塞做了处理，咱们来看下吧。

在RxJava2.0中官方，推出了Flowable 和Subscriber用来支持背压，同样的去除了Observable对背压的支持，对的就像你上面看到的，Observable不再支持背压了，即使阻塞崩溃也不会抛出MissingBackpressureException

还是上代码看看，Flowable的用法吧。

Flowable.create(FlowableOnSubscribe source, BackpressureStrategy mode)

创建Flowable会默认让传入一个FlowableOnSubscribe和一个BackpressureStrategy，FlowableOnSubscribe很好理解就是一个就是Flowable的一个被观察者源，而BackpressureStrategy就是Flowable提供的背压策略

有哪些策略还是上源码看下吧：

public enum BackpressureStrategy {
/**

• OnNext events are written without any buffering or dropping.
• Downstream has to deal with any overflow.

这篇好文章是转载于：学新通技术网