FlowableOperator 是否天生就支持背压？答案

【问题标题】：Does a FlowableOperator inherently supports backpressure?FlowableOperator 是否天生就支持背压？
【发布时间】：2018-02-22 14:37:19
【问题描述】：

我已经按照 RxJava2 wiki (https://github.com/ReactiveX/RxJava/wiki/Writing-operators-for-2.0#operator-targeting-lift) 中的描述实现了 FlowableOperator，只是我在 onNext() 操作中执行了一些类似这样的测试：

public final class MyOperator implements FlowableOperator<Integer, Integer> {

...

static final class Op implements FlowableSubscriber<Integer>, Subscription {

    @Override
    public void onNext(Integer v) {
        if (v % 2 == 0) {
          child.onNext(v * v);
        }  
    }
   ...
  }
}

这个运算符是一个链的一部分，我在其中创建了一个带有背压下降的Flowable。本质上，它看起来几乎是这样的：

Flowable.<Integer>create(emitter -> myAction(), DROP)
        .filter(v -> v > 2)
        .lift(new MyOperator())
        .subscribe(n -> doSomething(n));

我遇到了以下问题：

出现背压，因此doSomething(n) 无法处理即将到来的上游
由于选择了背压策略，项目被丢弃
但是 doSomething(n) 在执行 drop 之后永远不会收到新项目，而 doSomething(n) 已准备好处理新项目

回顾 David Karnok 的优秀博文 http://akarnokd.blogspot.fr/2015/05/pitfalls-of-operator-implementations.html，看来我需要在 onNext() 方法中添加 request(1)。但那是用 RxJava1...

所以，我的问题是：这个修复在 RxJava2 中是否足以解决我的背压问题？还是我的操作员必须实现所有关于 Atomics 的内容，排空https://github.com/ReactiveX/RxJava/wiki/Writing-operators-for-2.0#atomics-serialization-deferred-actions 中描述的内容才能正确处理我的背压问题？

注意：我添加了request(1)，它似乎可以工作。但我不知道这是否足够，或者我的操作员是否需要队列排水和原子等棘手的东西。

提前致谢！

【问题讨论】：

标签： rx-java2 backpressure

【解决方案1】：

FlowableOperator 是否天生就支持背压？

FlowableOperator 是一个为给定的下游Subscriber 调用的接口，并且应该返回一个新的Subscriber，它包装下游并调节在一个或两个方向上传递的 Reactive Streams 事件。背压支持是Subscriber 实现的责任，而不是这个特定的功能接口。它可能是Function<Subscriber, Subscriber>，但一个单独的命名接口被认为更有用，更不容易发生过载冲突。

需要在 onNext() [...] 中添加一个 request(1) 但我不知道这是否足够，或者我的操作员是否需要队列排水和原子等棘手的东西。

是的，你也必须在 RxJava 2 中这样做。由于 RxJava 2 的 Subscriber 不是一个类，它没有 v1 的方便 request 方法。您必须将Subscription 保存在onSubscribe 中，并在onNext 中的适当路径上调用upstream.request(1)。对于你的情况，应该足够了。

我已经更新了 wiki，新增了一个明确解释此案例的部分：

https://github.com/ReactiveX/RxJava/wiki/Writing-operators-for-2.0#replenishing

final class FilterOddSubscriber implements FlowableSubscriber<Integer>, Subscription {

    final Subscriber<? super Integer> downstream;

    Subscription upstream;

    // ...

    @Override
    public void onSubscribe(Subscription s) {
        if (upstream != null) {
            s.cancel();
        } else {
            upstream = s;                    // <-------------------------
            downstream.onSubscribe(this);
        }
    }

    @Override
    public void onNext(Integer item) {
        if (item % 2 != 0) {
           downstream.onNext(item);
        } else {
           upstream.request(1);              // <-------------------------
        }
    }

    @Override
    public void request(long n) {
        upstream.request(n);
    }

    // the rest omitted for brevity
}

【讨论】：

感谢您的回复和维基更新！因此，根据我的理解，添加upstream.request(1) 足以支持我的背压。如果下游请求超过 1 个元素是不够的。在这种情况下，必须实现 wiki 中描述的 drain-queue 东西。我说的对吗？
这取决于操作者的行为，而不是下游的请求模式。如果您丢弃 1 个项目，则必须在此中间运算符中向上游请求 1 个更多项目。当存在异步边界时，队列排水就会发挥作用，而不是在您的普通顺序运算符中。

【解决方案2】：

是的，你必须做一些棘手的事情......

我会避免编写运算符，除非您非常确定自己在做什么？几乎所有事情都可以使用默认操作符来实现...

编写运算符，类源 (fromEmitter) 或类中间 (flatMap) 在 RxJava 中一直是一项艰巨的任务。有许多要遵守的规则，要考虑许多情况，但同时要考虑许多情况（合法）构建性能良好的代码的捷径。现在写专门针对 2.x 的算子比 1.x 难 10 倍。如果您想利用所有先进的第 4 代功能，那就是甚至比顶部更难 2-3 倍（总共 30 倍）。

这里解释了一些棘手的东西：https://github.com/ReactiveX/RxJava/wiki/Writing-operators-for-2.0

【讨论】：