rx-java 介绍
示例
背压是在Observable处理管道中时，某些异步阶段无法足够快地处理值，因此需要一种方法来告诉上游生产者放慢速度。
需要背压的经典情况是生产者是热源时：
PublishSubject<Integer> source = PublishSubject.create();
source
.observeOn(Schedulers.computation())
.subscribe(v -> compute(v), Throwable::printStackTrace);
for (int i = 0; i < 1_000_000; i++) {
source.onNext(i);
}
Thread.sleep(10_000);在此示例中，主线程将向在后台线程上对其进行处理的最终消费者生产100万件商品。该方法可能会compute(int)花费一些时间，但Observable运算符链的开销也可能会增加处理项目的时间。但是，带有for循环的生产线程不知道这一点，并且会继续执行onNext。
在内部，异步运算符具有缓冲区来保存此类元素，直到可以对其进行处理为止。在经典Rx.NET和早期的RxJava中，这些缓冲区是无界的，这意味着它们可能会容纳该示例中几乎所有的100万个元素。例如，当程序中有10亿个元素或同一100万个序列出现1000次时，问题就开始出现，OutOfMemoryError并通常由于过度的GC开销而导致速度变慢。
与错误处理成为一流公民并通过onErrorXXX运算符（通过运算符）来处理错误的方式类似，背压是程序员必须考虑和处理（通过onBackpressureXXX运算符）的数据流的另一个属性。
除PublishSubject上述之外，还有其他一些不支持背压的运算符，主要是由于功能原因。例如，运算符interval会定期发出值，将其反向加压会导致相对于挂钟的时间发生偏移。
在现代RxJava中，大多数异步运算符现在都具有一个有界的内部缓冲区，就像observeOn上面一样，任何尝试溢出该缓冲区的操作都将使用终止整个序列MissingBackpressureException。每个运算符的文档中都有关于其反压行为的描述。
但是，背压在常规冷序中会更微妙地出现（不会也不应屈服MissingBackpressureException）。如果第一个示例被重写：
Observable.range(1, 1_000_000) .observeOn(Schedulers.computation()) .subscribe(v -> compute(v), Throwable::printStackTrace); Thread.sleep(10_000);
没有错误，并且使用少量内存即可使所有操作顺利进行。这样做的原因是，许多源运算符可以按需“生成”值，因此运算符observeOn可以告诉range生成的observeOn缓冲区最多可以一次拥有这么多的值，而不会溢出。
该协商基于协同例程的计算机科学概念（我叫你，你叫我）。运算符通过调用其（inner's）range以Producer接口实现的形式observeOn向其发送回调。作为回报，带有值的调用将告诉它允许产生（即，它）许多其他元素。然后，有责任在正确的时间以正确的值调用该方法，以保持数据畅通但不会溢出。SubscribersetProducerobserveOnProducer.request(n)rangeonNextobserveOnrequest
很少需要在最终用户中表达背压（因为它们相对于其直接上游是同步的，并且背压自然是由于调用堆栈阻塞而发生的），但可能更容易理解它的工作原理：
Observable.range(1, 1_000_000)
.subscribe(new Subscriber<Integer>() {
@Override
public void onStart() {
request(1);
}
public void onNext(Integer v) {
compute(v);
request(1);
}
@Override
public void onError(Throwable ex) {
ex.printStackTrace();
}
@Override
public void onCompleted() {
System.out.println("Done!");
}
});在此，onStart实现指示range生成其第一个值，然后在中接收该值onNext。一旦compute(int)完成，将另一个值，然后从请求range。在的天真的实现中range，此类调用将递归调用onNext，StackOverflowError这当然是不可取的。
为避免这种情况，运营商使用所谓的蹦床逻辑来防止此类可重入的呼叫。用range的术语，它将记住在request(1)调用时有一个调用onNext()，一旦onNext()返回，它将进行onNext()下一回合并使用下一个整数值进行调用。因此，如果将两者交换，该示例仍然可以正常工作：
@Override
public void onNext(Integer v) {
request(1);
compute(v);
}但是，这不适用于onStart。尽管Observable基础结构保证每次将最多Subscriber调用一次，但是对的调用request(1)可能会立即触发元素的发射。如果在request(1)需要调用之后具有初始化逻辑，则onNext可能会出现以下异常：
Observable.range(1, 1_000_000)
.subscribe(new Subscriber<Integer>() {
String name;
@Override
public void onStart() {
request(1);
name = "RangeExample";
}
@Override
public void onNext(Integer v) {
compute(name.length + v);
request(1);
}
//...休息是一样的
});在这种同步情况下，NullPointerException将在仍然执行时立即抛出aonStart。如果对的调用request(1)触发了onNext对其他线程的异步调用，并name在onNext竞赛中读取了将其写入onStart后的情况，则会发生更细微的错误request。
因此，应该在此onStart之前或之前进行所有字段初始化，然后调用request()last。request()在必要时，in运算符的实现可确保适当的事前发生关系（或换句话说，内存释放或完全隔离）。