如何在 Reactor 3 中编写运算符？

Question

我通过实现org.reactivestreams.Publisher 实现了一个 Reactor 运算符，如下所示。但是，我想知道这是否是正确的方式 ™ 使用 Reactor。手动实现订阅者看起来有点麻烦。而Operators 类似乎在这方面没有帮助。

class MyOperator implements Publisher<Integer> {

    private final Publisher<Integer> source;

    public MyOperator(Publisher<Integer> source) {
        this.source = source;
    }

    @Override
    public void subscribe(Subscriber<? super Integer> target) {
        final Subscriber<Integer> wrappedSubscriber = createSubscriber(target);
        source.subscribe(wrappedSubscriber);
    }

    private Subscriber<Integer> createSubscriber(Subscriber<? super Integer> target) {
        return new Subscriber<Integer>() {
            private Subscription subscription;

            @Override
            public void onSubscribe(Subscription subscription) {
                this.subscription = subscription;
                subscription.request(Long.MAX_VALUE);
            }

            @Override
            public void onNext(Integer integer) {
                target.onNext(integer + 1); // actual behaviour
                subscription.request(Long.MAX_VALUE);
            }

            @Override
            public void onError(Throwable t) {
                target.onError(t);
            }

            @Override
            public void onComplete() {
                target.onComplete();
            }
        };
    }
}

或者下面的例子是正确的方式™？

class MyCompactOperator implements Publisher<Integer> {

    final Flux<Integer> flux;

    public MyCompactOperator(Publisher<Integer> source) {
        flux = Flux.from(source).map(number -> number + 1);
    }

    @Override
    public void subscribe(Subscriber<? super Integer> subscriber) {
        flux.subscribe(subscriber);
    }
}

至少这需要更少的代码。

正如 Simon Baslé 所建议的那样，以 Flux 作为来源的变体 3：

class MyFluxOperator extends Flux<Integer> {

    private final Flux<Integer> source;

    public MyFluxOperator(Flux<Integer> source) {
        this.source = source;
    }

    @Override
    public void subscribe(Subscriber<? super Integer> subscriber) {
        source.map(number -> number + 1).subscribe(subscriber);
    }
}

所有实现都按预期工作：

Flux<Integer> source = Flux.just(1, 2, 3, 4, 5);
Flux.from(new MyOperator(source)).subscribe(System.out::println);

// for variant 3
new MyFluxOperator(source).subscribe(System.out::println);

我在第二行使用了Flux 来避免实现另一个订阅者。

输出：

2
3
4
5
6

问题：

• 我的实现中是否缺少某些内容？
• 有没有更好的方法（代码更少，错误处理更好等等）在 Reactor 3 中实现运算符？
• 这两种方法之间是否存在相关的功能或非功能差异？

Answer 1

看到您的第二个选项，您似乎认为您必须实现一个发布者。绝对不是这种情况（相反）。从 reactor Flux 源（或 Publisher + Reactor 的 Flux.from）开始，并简单地链接到 map。

编辑：为了澄清您不想创建任何类，只需在您的主代码路径中执行此操作：

• 如果您的source 已经是Flux 或Mono：

  Flux<Integer> incrementedSource = source.map(i -> i + 1);
  incrementedSource.subscribe(subscriber);
  

• 如果你的source是另一种Publisher：

  Flux<Integer> incrementedSource = Flux.from(source).map(i -> i + 1);
  incrementedSource.subscribe(subscriber);
  

像 Reactor 这样的库的整个想法是为您提供可以直接编写的运算符，而无需编写 Publisher

如果您因为经常将一组运算符应用于各种 Flux 而想要一种使代码相互化的方法，请查看 transform 和 compose（以及 reference documentation）。